/*
 * Copyright 2017-2022 EPAM Systems, Inc. (https://www.epam.com/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.epam.pipeline.manager.pipeline;

import com.epam.pipeline.acl.folder.FolderApiService;
import com.epam.pipeline.common.MessageConstants;
import com.epam.pipeline.common.MessageHelper;
import com.epam.pipeline.config.JsonMapper;
import com.epam.pipeline.controller.PagedResult;
import com.epam.pipeline.controller.vo.PagingRunFilterVO;
import com.epam.pipeline.controller.vo.PipelineRunFilterVO;
import com.epam.pipeline.controller.vo.PipelineRunServiceUrlVO;
import com.epam.pipeline.controller.vo.TagsVO;
import com.epam.pipeline.dao.pipeline.PipelineRunDao;
import com.epam.pipeline.entity.AbstractSecuredEntity;
import com.epam.pipeline.entity.BaseEntity;
import com.epam.pipeline.entity.cluster.InstanceDisk;
import com.epam.pipeline.entity.cluster.InstanceOffer;
import com.epam.pipeline.entity.cluster.InstancePrice;
import com.epam.pipeline.entity.cluster.PriceType;
import com.epam.pipeline.entity.configuration.ExecutionEnvironment;
import com.epam.pipeline.entity.configuration.PipeConfValueVO;
import com.epam.pipeline.entity.configuration.PipelineConfiguration;
import com.epam.pipeline.entity.contextual.ContextualPreferenceExternalResource;
import com.epam.pipeline.entity.contextual.ContextualPreferenceLevel;
import com.epam.pipeline.entity.datastorage.AbstractDataStorage;
import com.epam.pipeline.entity.docker.ToolVersion;
import com.epam.pipeline.entity.metadata.MetadataEntity;
import com.epam.pipeline.entity.metadata.PipeConfValue;
import com.epam.pipeline.entity.metadata.PipeConfValueType;
import com.epam.pipeline.entity.metadata.RunStatusMetadata;
import com.epam.pipeline.entity.pipeline.CommitStatus;
import com.epam.pipeline.entity.pipeline.DiskAttachRequest;
import com.epam.pipeline.entity.pipeline.DockerRegistry;
import com.epam.pipeline.entity.pipeline.Folder;
import com.epam.pipeline.entity.pipeline.Pipeline;
import com.epam.pipeline.entity.pipeline.PipelineRun;
import com.epam.pipeline.entity.pipeline.PipelineRunWithTool;
import com.epam.pipeline.entity.pipeline.RunInstance;
import com.epam.pipeline.entity.pipeline.TaskStatus;
import com.epam.pipeline.entity.pipeline.Tool;
import com.epam.pipeline.entity.pipeline.run.ExecutionPreferences;
import com.epam.pipeline.entity.pipeline.run.PipeRunCmdStartVO;
import com.epam.pipeline.entity.pipeline.run.PipelineStart;
import com.epam.pipeline.entity.pipeline.run.PipelineStartNotificationRequest;
import com.epam.pipeline.entity.pipeline.run.RestartRun;
import com.epam.pipeline.entity.pipeline.run.RunAssignPolicy;
import com.epam.pipeline.entity.pipeline.run.RunInfo;
import com.epam.pipeline.entity.pipeline.run.RunStatus;
import com.epam.pipeline.entity.pipeline.run.parameter.PipelineRunParameter;
import com.epam.pipeline.entity.pipeline.run.parameter.RunSid;
import com.epam.pipeline.entity.region.AbstractCloudRegion;
import com.epam.pipeline.entity.security.acl.AclClass;
import com.epam.pipeline.entity.user.PipelineUser;
import com.epam.pipeline.entity.utils.DateUtils;
import com.epam.pipeline.exception.git.GitClientException;
import com.epam.pipeline.manager.cluster.InstanceOfferManager;
import com.epam.pipeline.manager.cluster.KubernetesConstants;
import com.epam.pipeline.manager.cluster.NodesManager;
import com.epam.pipeline.manager.cluster.pool.NodePoolManager;
import com.epam.pipeline.manager.datastorage.DataStorageManager;
import com.epam.pipeline.manager.docker.DockerRegistryManager;
import com.epam.pipeline.manager.docker.scan.ToolSecurityPolicyCheck;
import com.epam.pipeline.manager.execution.PipelineLauncher;
import com.epam.pipeline.manager.git.GitManager;
import com.epam.pipeline.manager.metadata.MetadataEntityManager;
import com.epam.pipeline.manager.notification.ContextualNotificationRegistrationManager;
import com.epam.pipeline.manager.pipeline.runner.ConfigurationProviderManager;
import com.epam.pipeline.manager.pipeline.runner.PipeRunCmdBuilder;
import com.epam.pipeline.manager.preference.PreferenceManager;
import com.epam.pipeline.manager.preference.SystemPreferences;
import com.epam.pipeline.manager.quota.RunLimitsService;
import com.epam.pipeline.manager.region.CloudRegionManager;
import com.epam.pipeline.manager.security.AuthManager;
import com.epam.pipeline.manager.security.CheckPermissionHelper;
import com.epam.pipeline.manager.security.run.RunPermissionManager;
import com.epam.pipeline.utils.CommonUtils;
import com.epam.pipeline.utils.PasswordGenerator;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.epam.pipeline.entity.configuration.RunConfigurationUtils.getNodeCount;
import static com.epam.pipeline.manager.pipeline.ToolUtils.REPOSITORY_AND_IMAGE;
import static com.epam.pipeline.manager.pipeline.ToolUtils.getImageWithoutTag;
import static java.util.stream.Collectors.mapping;
import static java.util.stream.Collectors.toSet;

@Slf4j
@Service
public class PipelineRunManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(PipelineRunManager.class);

    private static final int POD_ID_LENGTH = 63;
    private static final String EMPTY_STRING = "";
    private static final Long EMPTY_PIPELINE_ID = null;
    private static final int MILLS_IN_SEC = 1000;
    private static final int DIVIDER_TO_GB = 1024 * 1024 * 1024;
    private static final int USER_PRICE_SCALE = 2;
    private static final int BILLING_PRICE_SCALE = 5;
    public static final String CP_CAP_LIMIT_MOUNTS = "CP_CAP_LIMIT_MOUNTS";
    private static final String LIMIT_MOUNTS_NONE = "none";
    private static final String CP_REPORT_RUN_STATUS = "CP_REPORT_RUN_STATUS";
    private static final String CP_REPORT_RUN_PROCESSED_DATE = "CP_REPORT_RUN_PROCESSED_DATE";
    private static final String CP_GPU_COUNT = "CP_GPU_COUNT";

    @Autowired
    private PipelineRunDao pipelineRunDao;

    @Autowired
    private PipelineManager pipelineManager;

    @Autowired
    private PipelineVersionManager versionManager;

    @Autowired
    private MessageHelper messageHelper;

    @Autowired
    private PipelineLauncher pipelineLauncher;

    @Autowired
    private GitManager gitManager;

    @Autowired
    private DataStorageManager dataStorageManager;

    @Autowired
    private InstanceOfferManager instanceOfferManager;

    @Autowired
    private ToolManager toolManager;

    @Autowired
    private RunPermissionManager permissionManager;

    @Autowired
    private PipelineConfigurationManager configurationManager;

    @Autowired
    private FolderApiService folderApiService;

    @Autowired
    private PreferenceManager preferenceManager;

    @Autowired
    private CloudRegionManager cloudRegionManager;

    @Autowired
    private AuthManager authManager;

    @Autowired
    private ConfigurationProviderManager configurationProviderManager;

    @Autowired
    private RestartRunManager restartRunManager;

    @Autowired
    private RunStatusManager runStatusManager;

    @Autowired
    private NodesManager nodesManager;
    
    @Autowired
    private CheckPermissionHelper permissionHelper;

    @Autowired
    private PipelineRunCRUDService runCRUDService;

    @Autowired
    private StopServerlessRunManager stopServerlessRunManager;

    @Autowired
    private DockerRegistryManager dockerRegistryManager;

    @Autowired
    private PipelineRunServiceUrlManager pipelineRunServiceUrlManager;

    @Autowired
    private ContextualNotificationRegistrationManager contextualNotificationRegistrationManager;

    @Autowired
    private NodePoolManager nodePoolManager;

    @Autowired
    private RunLimitsService runLimitsService;

    @Autowired
    private MetadataEntityManager metadataEntityManager;

    /**
     * Launches cmd command execution, uses Tool as ACL identity
     * @param runVO
     * @return
     */
    @ToolSecurityPolicyCheck
    public PipelineRun runCmd(final PipelineStart runVO) {
        LOGGER.debug("Tool '{}' will be launched on behalf of '{}'", runVO.getDockerImage(),
                authManager.getAuthorizedUser());
        Assert.notNull(runVO.getInstanceType(),
                messageHelper.getMessage(MessageConstants.SETTING_IS_NOT_PROVIDED, "instance_type"));
        Assert.notNull(runVO.getHddSize(),
                messageHelper.getMessage(MessageConstants.SETTING_IS_NOT_PROVIDED, "instance_disk"));

        final int maxRunsNumber = preferenceManager.getPreference(SystemPreferences.LAUNCH_MAX_SCHEDULED_NUMBER);

        LOGGER.debug("Allowed runs count - {}, actual - {}", maxRunsNumber, getNodeCount(runVO.getNodeCount(), 1));
        Assert.isTrue(getNodeCount(runVO.getNodeCount(), 1) <= maxRunsNumber, messageHelper.getMessage(
                MessageConstants.ERROR_EXCEED_MAX_RUNS_COUNT, maxRunsNumber, getNodeCount(runVO.getNodeCount(), 1)));
        checkRunLaunchLimits(runVO);
        final Tool tool = toolManager.loadByNameOrId(runVO.getDockerImage());
        final PipelineConfiguration configuration = configurationManager.getPipelineConfiguration(runVO, tool);
        runVO.setRunSids(configuration.mergeRunSids(runVO.getRunSids()));
        final boolean clusterRun = configurationManager.initClusterConfiguration(configuration, true);

        final PipelineRun run = launchPipeline(configuration, null, null,
                runVO.getInstanceType(), runVO.getConfigurationName(), null,
                runVO.getParentRunId(), null, null, runVO.getRunSids(),
                configuration.getNotifications());
        run.setParent(tool);
        run.setAclClass(AclClass.TOOL);

        if (clusterRun) {
            runClusterWorkers(run, runVO, null, null, configuration);
        }
        return run;
    }

    /**
     * Creates a new pod with a given run_id, doesn't create a new pipeline run
     * @param runVO
     * @return
     */
    //TODO: refactoring
    @ToolSecurityPolicyCheck
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun runPod(final PipelineStart runVO) {
        Assert.notNull(runVO.getCmdTemplate(),
                messageHelper.getMessage(MessageConstants.SETTING_IS_NOT_PROVIDED, "cmd_template"));
        final PipelineRun parentRun = loadPipelineRun(runVO.getUseRunId(), false);
        Assert.state(parentRun.getStatus() == TaskStatus.RUNNING,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_RUN_NOT_RUNNING, runVO.getUseRunId()));
        checkRunLaunchLimits(runVO);
        final PipelineConfiguration configuration = configurationManager.getPipelineConfiguration(runVO);
        final Tool tool = getToolForRun(configuration);
        configuration.setSecretName(tool.getSecretName());
        final List<String> endpoints = tool.getEndpoints();
        final PipelineRun run = new PipelineRun();
        run.setInstance(parentRun.getInstance());
        run.setId(runVO.getUseRunId());
        run.setStartDate(DateUtils.now());
        run.setStatus(TaskStatus.RUNNING);
        run.setPipelineName(PipelineRun.DEFAULT_PIPELINE_NAME);
        run.setPodId(getRootPodIDFromTool(tool.getImage(), run.getId()));
        run.setDockerImage(configuration.getDockerImage());
        run.setActualDockerImage(tool.getImage());
        run.setCmdTemplate(determinateCmdTemplateForRun(configuration));
        run.setTimeout(runVO.getTimeout());
        run.setCommitStatus(CommitStatus.NOT_COMMITTED);
        run.setLastChangeCommitTime(DateUtils.now());
        run.setRunSids(runVO.getRunSids());
        run.setOwner(parentRun.getOwner());
        final String launchedCommand = pipelineLauncher.launch(
                run, configuration, endpoints, false, parentRun.getPodId(), null
        );
        run.setActualCmd(launchedCommand);
        return run;
    }

    /**
     * Runs specified pipeline version, uses Pipeline as ACL identity
     *
     * @param runVO
     * @return
     */
    @ToolSecurityPolicyCheck
    public PipelineRun runPipeline(final PipelineStart runVO) {
        final Long pipelineId = runVO.getPipelineId();
        LOGGER.debug("Pipeline '{}' will be launched on behalf of '{}'", pipelineId, authManager.getAuthorizedUser());
        final String version = runVO.getVersion();
        final int maxRunsNumber = preferenceManager.getPreference(SystemPreferences.LAUNCH_MAX_SCHEDULED_NUMBER);

        LOGGER.debug("Allowed runs count - {}, actual - {}", maxRunsNumber, getNodeCount(runVO.getNodeCount(), 1));
        Assert.isTrue(getNodeCount(runVO.getNodeCount(), 1) <= maxRunsNumber, messageHelper.getMessage(
                MessageConstants.ERROR_EXCEED_MAX_RUNS_COUNT, maxRunsNumber, getNodeCount(runVO.getNodeCount(), 1)));
        checkRunLaunchLimits(runVO);
        final Pipeline pipeline = pipelineManager.load(pipelineId);
        final PipelineConfiguration configuration = configurationManager
                .getPipelineConfigurationForPipeline(pipeline, runVO);
        runVO.setRunSids(configuration.mergeRunSids(runVO.getRunSids()));
        final boolean isClusterRun = configurationManager.initClusterConfiguration(configuration, true);

        permissionManager.checkToolRunPermission(configuration.getDockerImage());
        final PipelineRun run = launchPipeline(configuration, pipeline, version,
                runVO.getInstanceType(), runVO.getConfigurationName(), null,
                runVO.getParentRunId(), null, null, runVO.getRunSids(),
                configuration.getNotifications());
        run.setParent(pipeline);

        if (isClusterRun) {
            runClusterWorkers(run, runVO, version, pipeline, configuration);
        }
        return run;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void prolongIdleRun(Long runId) {
        PipelineRun run = loadPipelineRun(runId, false);
        run.setLastIdleNotificationTime(null);
        run.setLastNotificationTime(null);
        run.setProlongedAtTime(DateUtils.nowUTC());
        updateProlongIdleRunAndLastIdleNotificationTime(run);
    }


    /**
     * Internal method for creating a pipeline run,
     * it assumes that ACL filtering was already applied to input arguments
     * @param configuration
     * @param pipeline - null in case of CMD launch
     * @param version - null in case of CMD launch
     * @param instanceType
     * @param configurationName
     * @param runSids - a list of identities (user names or groups) that have access to run
     * @return
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun launchPipeline(final PipelineConfiguration configuration, final Pipeline pipeline,
                                      final String version, final String instanceType, final String configurationName,
                                      final String clusterId, final Long parentRunId, final List<Long> entityIds,
                                      final Long configurationId, final List<RunSid> runSids,
                                      final List<PipelineStartNotificationRequest> notificationRequests) {
        final Optional<PipelineRun> parentRun = resolveParentRun(parentRunId, configuration);
        final Tool tool = getToolForRun(configuration);
        final Optional<ToolVersion> toolVersion = toolManager.findToolVersion(tool);
        final PipelineConfiguration toolConfiguration = configurationManager
                .getConfigurationForTool(tool, configuration);
        final AbstractCloudRegion region = resolveCloudRegion(
                parentRun.orElse(null), configuration, toolConfiguration);
        validateCloudRegion(toolConfiguration, region);
        validateInstanceAndPriceTypes(configuration, pipeline, region, instanceType);
        final String instanceDisk = configuration.getInstanceDisk();
        if (StringUtils.isNotBlank(instanceDisk)) {
            Assert.isTrue(NumberUtils.isNumber(instanceDisk) &&
                Integer.parseInt(instanceDisk) > 0,
                    messageHelper.getMessage(MessageConstants.ERROR_INSTANCE_DISK_IS_INVALID, instanceDisk));
        }

        adjustInstanceDisk(configuration);
        checkGPUInstance(configuration, region.getId());

        final List<String> endpoints = configuration.isEraseRunEndpoints()
                ? Collections.emptyList() : tool.getEndpoints();
        configuration.setSecretName(tool.getSecretName());
        final boolean sensitive = checkRunForSensitivity(configuration.getParameters());
        Assert.isTrue(!sensitive || tool.isAllowSensitive(),
                messageHelper.getMessage(
                        MessageConstants.ERROR_SENSITIVE_RUN_NOT_ALLOWED_FOR_TOOL, tool.getImage()));

        final PipelineRun run = createPipelineRun(version, configuration, pipeline, tool, toolVersion.orElse(null),
                region, parentRun.orElse(null), entityIds, configurationId, sensitive);

        // If there is no podAssignPolicy, then run is scheduled to a dedicated node
        if (configuration.getPodAssignPolicy() == null || !configuration.getPodAssignPolicy().isValid()) {
            log.debug(String.format("Configuring default pod assign policy for run #%d", run.getId()));
            configuration.setPodAssignPolicy(getDefaultPodAssignPolicy(run));
        }

        configuration.getPodAssignPolicy()
                .ifMatchThenMapValue(KubernetesConstants.RUN_ID_LABEL, Long::parseLong)
                .ifPresent(parentNodeId -> {
                    if (!parentNodeId.equals(run.getId())) {
                        setParentInstance(run, parentNodeId);
                    }
                });

        run.setConfigName(configurationName);
        run.setRunSids(runSids);
        final String launchedCommand = pipelineLauncher.launch(run, configuration, endpoints, clusterId);
        //update instance info according to evaluated command
        run.setActualCmd(launchedCommand);
        save(run);
        dataStorageManager.analyzePipelineRunsParameters(Collections.singletonList(run));
        contextualNotificationRegistrationManager.register(notificationRequests, run);
        if (MapUtils.isNotEmpty(configuration.getTags())) {
            run.setTags(configuration.getTags());
            pipelineRunDao.updateRunTags(run);
        }
        run.parseParameters();
        updateMetadataEntities(run);
        return run;
    }

    private void checkGPUInstance(final PipelineConfiguration configuration, final Long regionId) {
        final String instanceType = configuration.getInstanceType();
        if (StringUtils.isNotBlank(instanceType)) {
            final Optional<InstanceOffer> instance = instanceOfferManager.findOffer(instanceType, regionId);
            if (instance.isPresent()) {
                final InstanceOffer offer = instance.get();
                if (offer.getGpu() > 0) {
                    configuration.setParameters(CommonUtils.mergeMaps(
                            Collections.singletonMap(CP_GPU_COUNT,
                                    new PipeConfValueVO(String.valueOf(offer.getGpu()))),
                            configuration.getParameters()));
                    return;
                }
            }
        }
        MapUtils.emptyIfNull(configuration.getParameters()).remove(CP_GPU_COUNT);
    }

    private RunAssignPolicy getDefaultPodAssignPolicy(final PipelineRun run) {
        return RunAssignPolicy.builder()
                .selector(
                        RunAssignPolicy.PodAssignSelector.builder()
                                .label(KubernetesConstants.RUN_ID_LABEL)
                                .value(run.getId().toString()).build())
                .tolerances(Arrays.asList(
                        RunAssignPolicy.PodAssignTolerance.builder()
                                .label(KubernetesConstants.KUBE_UNREACHABLE_NODE_LABEL)
                                .value(StringUtils.EMPTY)
                                .build(),
                        RunAssignPolicy.PodAssignTolerance.builder()
                                .label(KubernetesConstants.KUBE_NOT_READY_NODE_LABEL)
                                .value(StringUtils.EMPTY)
                                .build()))
                .build();
    }

    private AbstractCloudRegion resolveCloudRegion(final PipelineRun parentRun,
                                                   final PipelineConfiguration configuration, 
                                                   final PipelineConfiguration toolConfiguration) {
        final Optional<Long> configurationRegionId = Optional.ofNullable(configuration)
                .map(PipelineConfiguration::getCloudRegionId);
        final Optional<Long> toolConfigurationRegionId = Optional.ofNullable(toolConfiguration)
                .map(PipelineConfiguration::getCloudRegionId);
        final Optional<Long> parentRunRegionId = Optional.ofNullable(parentRun)
                .map(PipelineRun::getInstance)
                .map(RunInstance::getCloudRegionId);
        return configurationRegionId.map(Optional::of)
                .orElse(parentRunRegionId).map(Optional::of)
                .orElse(toolConfigurationRegionId)
                .map(cloudRegionManager::load)
                .orElseGet(cloudRegionManager::loadDefaultRegion);
    }

    private void validateCloudRegion(final PipelineConfiguration toolConfiguration,
                                     final AbstractCloudRegion region) {
        if (!permissionHelper.isAdmin()) {
            if (toolConfiguration.getCloudRegionId() != null) {
                Assert.isTrue(toolConfiguration.getCloudRegionId().equals(region.getId()),
                        messageHelper.getMessage(MessageConstants.ERROR_TOOL_CLOUD_REGION_NOT_ALLOWED,
                                cloudRegionManager.load(toolConfiguration.getCloudRegionId()).getName(), 
                                region.getName()));
            }
            Assert.isTrue(permissionHelper.isAllowed("READ", region),
                    messageHelper.getMessage(MessageConstants.ERROR_RUN_CLOUD_REGION_NOT_ALLOWED, region.getName()));
        }
    }

    private void validateInstanceAndPriceTypes(final PipelineConfiguration configuration,
                                               final Pipeline pipeline,
                                               final AbstractCloudRegion region,
                                               final String instanceType) {
        final PriceType priceType = configuration.getIsSpot() != null && configuration.getIsSpot()
                ? PriceType.SPOT
                : PriceType.ON_DEMAND;
        final boolean isMaster = PipelineConfigurationManager.isClusterConfiguration(configuration);
        if (pipeline != null) {
            validatePipelineInstanceAndPriceTypes(instanceType, priceType, region.getId(), isMaster);
        } else {
            validateToolInstanceAndPriceTypes(instanceType, priceType,  region.getId(), configuration.getDockerImage(),
                                              isMaster);
        }
    }

    private void validatePipelineInstanceAndPriceTypes(final String instanceType,
                                                       final PriceType priceType,
                                                       final Long regionId,
                                                       final boolean isMasterNode) {
        Assert.isTrue(StringUtils.isBlank(instanceType)
                        || instanceOfferManager.isInstanceAllowed(instanceType, regionId, priceType == PriceType.SPOT),
                messageHelper.getMessage(MessageConstants.ERROR_INSTANCE_TYPE_IS_NOT_ALLOWED, instanceType));
        Assert.isTrue(instanceOfferManager.isPriceTypeAllowed(priceType.getLiteral(), null, isMasterNode),
                messageHelper.getMessage(MessageConstants.ERROR_PRICE_TYPE_IS_NOT_ALLOWED, priceType));
    }

    private void validateToolInstanceAndPriceTypes(final String instanceType,
                                                   final PriceType priceType,
                                                   final Long regionId,
                                                   final String dockerImage,
                                                   final boolean isMasterNode) {
        final Tool tool = toolManager.loadByNameOrId(dockerImage);
        final ContextualPreferenceExternalResource toolResource =
                new ContextualPreferenceExternalResource(ContextualPreferenceLevel.TOOL, tool.getId().toString());
        Assert.isTrue(StringUtils.isBlank(instanceType)
                        || instanceOfferManager.isToolInstanceAllowed(instanceType, toolResource,
                                                    regionId, priceType == PriceType.SPOT),
                messageHelper.getMessage(MessageConstants.ERROR_INSTANCE_TYPE_IS_NOT_ALLOWED, instanceType));
        Assert.isTrue(instanceOfferManager.isPriceTypeAllowed(priceType.getLiteral(), toolResource, isMasterNode),
                messageHelper.getMessage(MessageConstants.ERROR_PRICE_TYPE_IS_NOT_ALLOWED, priceType));
    }

    private Optional<PipelineRun> resolveParentRun(final Long parentRunId, final PipelineConfiguration configuration) {
        return resolveParentRunId(parentRunId, configuration).map(id -> loadPipelineRun(id, false));
    }

    private Optional<Long> resolveParentRunId(final Long parentRunId, final PipelineConfiguration configuration) {
        return Optional.ofNullable(parentRunId).map(Optional::of)
                .orElseGet(() -> resolveParentRunIdFromConfiguration(configuration));
    }

    private Optional<Long> resolveParentRunIdFromConfiguration(final PipelineConfiguration configuration) {
        return Optional.ofNullable(configuration)
                .map(PipelineConfiguration::getParameters)
                .filter(MapUtils::isNotEmpty)
                .map(map -> map.get(PipelineRun.PARENT_ID_PARAM))
                .map(PipeConfValueVO::getValue)
                .filter(NumberUtils::isDigits)
                .map(Long::parseLong);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public PipelineRun loadPipelineRun(Long id) {
        return loadPipelineRun(id, true);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public PipelineRun loadPipelineRun(Long id, boolean withLinks) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(id);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, id));
        setParent(pipelineRun);
        checkCommitRunStatus(pipelineRun);
        if (permissionManager.isRunSshAllowed(pipelineRun)) {
            pipelineRun.setSshPassword(pipelineRunDao.loadSshPassword(id));
        }
        pipelineRun.setPipelineRunParameters(
                replaceParametersWithEnvVars(pipelineRun.getPipelineRunParameters(), pipelineRun.getEnvVars()));
        if (withLinks) {
            dataStorageManager.analyzePipelineRunsParameters(Collections.singletonList(pipelineRun));
        }
        return pipelineRun;
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public Optional<PipelineRun> findRun(final Long id) {
        return Optional.ofNullable(pipelineRunDao.loadPipelineRun(id));
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public AbstractSecuredEntity loadRunParent(Long runId) {
        PipelineRun run = loadPipelineRun(runId, false);
        return loadRunParent(run);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public AbstractSecuredEntity loadRunParent(PipelineRun run) {
        if (run.getPipelineId() != null && run.getPipelineId() != 0) {
            return pipelineManager.load(run.getPipelineId());
        } else {
            return null;
        }
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public List<PipelineRun> loadPipelineRuns(List<Long> ids) {
        List<PipelineRun> pipelineRuns = pipelineRunDao.loadPipelineRuns(ids);
        Assert.notNull(pipelineRuns,
                messageHelper.getMessage(MessageConstants.ERROR_RUN_PIPELINES_NOT_FOUND,
                        ids.stream().map(Object::toString).collect(Collectors.joining(", "))));
        return pipelineRuns;
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public PipelineConfiguration loadRunConfiguration(Long id) throws GitClientException {
        PipelineRun pipelineRun = loadPipelineRun(id, false);
        return versionManager
                .loadParametersFromScript(pipelineRun.getPipelineId(),
                        pipelineRun.getVersion(), pipelineRun.getConfigName());
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateRunInstance(Long id, RunInstance instance) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(id);
        if (!instance.isEmpty()) {
            pipelineRun.setInstance(instance);
            pipelineRunDao.updateRunInstance(pipelineRun);
        }
        return pipelineRun;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updateRunInstanceStartDate(Long id, LocalDateTime date) {
        pipelineRunDao.updateRunInstanceStartDate(id, date);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePipelineStatusIfNotFinalExternal(Long runId, TaskStatus status) {
        return updatePipelineStatusIfNotFinal(runId, status);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePipelineStatusIfNotFinal(Long runId, TaskStatus status) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        if (pipelineRun.getStatus().isFinal()) {
            LOGGER.debug("Pipeline run {} is already in the final status: {}.",
                    pipelineRun.getId(), pipelineRun.getStatus());
            return pipelineRun;
        }
        if (pipelineRun.getExecutionPreferences().getEnvironment() == ExecutionEnvironment.DTS
                && status == TaskStatus.STOPPED) {
            configurationProviderManager.stop(runId, pipelineRun.getExecutionPreferences());
        }
        return updatePipelineStatus(runId, status, pipelineRun);
    }

    @SuppressWarnings("PMD.AvoidCatchingGenericException")
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePipelineStatus(final PipelineRun run) {
        try {
            updateMetadataEntities(run);
        } catch (Exception e) {
            LOGGER.error("An error occurred during metadata run status processing", e);
        }
        return runCRUDService.updateRunStatus(run);
    }

    /**
     * A shorthand method to stop a Pipeline Run
     * @param runId ID of a Pipeline Run
     * @return stopped run
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun stop(Long runId) {
        return updatePipelineStatusIfNotFinal(runId, TaskStatus.STOPPED);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePipelineCommitStatus(PipelineRun run) {
        pipelineRunDao.updateRunCommitStatus(run);
        return run;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePodIP(PipelineRun run) {
        pipelineRunDao.updatePodIP(run);
        return run;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePipelineRunLastNotification(PipelineRun run) {
        pipelineRunDao.updateRunLastNotification(run);
        return run;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updatePipelineRunsLastNotification(Collection<PipelineRun> runs) {
        pipelineRunDao.updateRunsLastNotification(runs);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateProlongIdleRunAndLastIdleNotificationTime(PipelineRun run) {
        pipelineRunDao.updateProlongIdleRunAndLastIdleNotificationTime(run);
        return run;
    }

    public List<PipelineRun> loadAllRunsByPipeline(Long id) {
        pipelineManager.load(id);
        return pipelineRunDao.loadAllRunsForPipeline(id);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public List<PipelineRun> loadAllRunsByPipeline(Long pipelineId, String version) {
        return pipelineRunDao.loadAllRunsForPipeline(pipelineId, version);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public List<PipelineRun> loadRunningAndTerminatedPipelineRuns() {
        return pipelineRunDao.loadRunningAndTerminatedPipelineRuns();
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public List<PipelineRun> loadRunningPipelineRuns() {
        return pipelineRunDao.loadRunningPipelineRuns();
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun loadPipelineRunWithRestartedRuns(Long id) {
        PipelineRun run = loadPipelineRun(id);
        List<RestartRun> restartedRuns = restartRunManager.loadRestartedRunsForInitialRun(id);
        if (CollectionUtils.isNotEmpty(restartedRuns)) {
            addRegionsToRestartedRuns(restartedRuns);
        }
        run.setRestartedRuns(restartedRuns);
        run.setRunStatuses(runStatusManager.loadRunStatus(id));
        return run;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun loadPipelineRunWithStatuses(final Long id) {
        final PipelineRun run = loadPipelineRun(id);
        run.setRunStatuses(runStatusManager.loadRunStatus(id));
        return run;
    }

    /**
     * Method that will return all active runs for which current users is owner or is listed in run sids - a list of
     * identities (user names or groups) that have access to run
     * @param filter - filter containing a page and page size
     * @return list of active runs which available for current user
     */
    @Transactional(propagation = Propagation.SUPPORTS)
    public PagedResult<List<PipelineRun>> loadActiveSharedRuns(PagingRunFilterVO filter) {
        Assert.isTrue(filter.getPage() > 0,
                messageHelper.getMessage(MessageConstants.ERROR_PAGE_INDEX));
        Assert.isTrue(filter.getPageSize() > 0,
                messageHelper.getMessage(MessageConstants.ERROR_PAGE_SIZE));
        PipelineUser user = authManager.getCurrentUser();
        List<PipelineRun> runs = pipelineRunDao.loadActiveSharedRuns(filter, user);
        int count = pipelineRunDao.countActiveSharedRuns(user);
        return new PagedResult<>(runs, count);
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public PagedResult<List<PipelineRun>> searchPipelineRuns(PagingRunFilterVO filter, boolean loadStorageLinks) {
        Assert.isTrue(filter.getPage() > 0,
                messageHelper.getMessage(MessageConstants.ERROR_PAGE_INDEX));
        Assert.isTrue(filter.getPageSize() > 0,
                messageHelper.getMessage(MessageConstants.ERROR_PAGE_SIZE));
        PipelineRunFilterVO.ProjectFilter projectFilter = resolveProjectFiltering(filter);
        if (projectFilter!= null && projectFilter.isEmpty()) {
            return new PagedResult<>(Collections.emptyList(), 0);
        }
        PagedResult<List<PipelineRun>> result;
        // todo: Use grouping request if parentId is specified. Otherwise use non grouping request.
        //  Non grouping requests return both parent and child runs in plain list (no nested runs).
        //  Grouping requests return only parent runs.
        // todo: Get rid of userModified and eagerGrouping flags
        if (filter.useGrouping()) {
            if (filter.isEagerGrouping()) {
                result = eagerSearchParentRuns(filter, projectFilter);
            } else {
                result = searchParentRuns(filter, projectFilter);
            }
        } else {
            result = searchRuns(filter, projectFilter);
        }
        if (loadStorageLinks && CollectionUtils.isNotEmpty(result.getElements())) {
            dataStorageManager.analyzePipelineRunsParameters(result.getElements());
        }
        if (CollectionUtils.isNotEmpty(result.getElements())) {
            Map<Long, List<RunStatus>> runStatuses = runStatusManager.loadRunStatus(result.getElements().stream()
                    .map(PipelineRun::getId).collect(Collectors.toList()));
            for (final PipelineRun run : result.getElements()) {
                run.setRunStatuses(runStatuses.get(run.getId()));
            }
        }
        return result;
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public Integer countPipelineRuns(PipelineRunFilterVO filter) {
        PipelineRunFilterVO.ProjectFilter projectFilter = resolveProjectFiltering(filter);
        if (projectFilter!= null && projectFilter.isEmpty()) {
            return 0;
        }
        return filter.useGrouping() ?
                pipelineRunDao.countRootRuns(filter, projectFilter) :
                pipelineRunDao.countFilteredPipelineRuns(filter, projectFilter);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateServiceUrl(final Long runId, final String region,
                                        final PipelineRunServiceUrlVO serviceUrl) {
        final PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(pipelineRun, messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, runId));

        pipelineRunServiceUrlManager.update(runId, region, serviceUrl);
        pipelineRun.setServiceUrl(pipelineRunServiceUrlManager.loadByRunId(runId));

        //TODO: check if we need it
        setParent(pipelineRun);
        return pipelineRun;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePrettyUrl(Long runId, String url) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, runId));
        if (Objects.equals(pipelineRun.getPrettyUrl(), url)) {
            LOGGER.debug("Url '{}' is already assigned to run {}", url, runId);
            return pipelineRun;
        }
        validatePrettyUrlFree(url);
        pipelineRun.setPrettyUrl(url);
        pipelineRunDao.updateRun(pipelineRun);
        //TODO: check if we need it
        setParent(pipelineRun);
        return pipelineRun;
    }

    public PipelineRun loadRunByPrettyUrl(final String url) {
        final PagingRunFilterVO filter = new PagingRunFilterVO();
        filter.setPage(1);
        filter.setPageSize(1);
        filter.setPrettyUrl(url);
        final PipelineRun run = pipelineRunDao.searchPipelineRuns(filter, null).stream().findFirst()
                .orElseThrow(() ->
                        new IllegalArgumentException(
                                messageHelper.getMessage(MessageConstants.ERROR_RUN_PRETTY_NOT_FOUND, url)));
        pipelineRunDao.loadRunFields(run);
        if (permissionManager.isRunSshAllowed(run)) {
            run.setSshPassword(pipelineRunDao.loadSshPassword(run.getId()));
        }
        return run;
    }

    /**
     * Commits docker image and push it to a docker registry from specified run
     * @param id {@link PipelineRun} id for pipeline run which commit status should be updated
     * @param commitStatus new {@link CommitStatus} of the pipeline run
     * @return updated {@link PipelineRun}
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateCommitRunStatus(Long id, CommitStatus commitStatus) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(id);
        toolManager.validateCommitOperationAllowed(pipelineRun.getActualDockerImage());

        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, id));
        pipelineRun.setCommitStatus(commitStatus);
        pipelineRun.setLastChangeCommitTime(DateUtils.now());
        return updatePipelineCommitStatus(pipelineRun);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updatePodStatus(Long id, String status) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(id);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, id));
        pipelineRun.setPodStatus(status);
        pipelineRunDao.updatePodStatus(pipelineRun);
        return pipelineRun;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun createPipelineRun(String version, PipelineConfiguration configuration, Pipeline pipeline,
                                         Tool tool, ToolVersion toolVersion, AbstractCloudRegion region, 
                                         PipelineRun parentRun, List<Long> entityIds, Long configurationId, 
                                         boolean sensitive) {
        validateRunParameters(configuration, pipeline);

        RunInstance instance = configureRunInstance(configuration, region);

        PipelineRun run = new PipelineRun();
        Long runId = pipelineRunDao.createRunId();
        run.setId(runId);
        run.setStartDate(DateUtils.now());
        run.setProlongedAtTime(DateUtils.nowUTC());
        if (pipeline == null || version == null) {
            fillMissingPipelineFields(run);
        } else {
            run.setPipelineName(pipeline.getName());
            run.setRepository(pipeline.getRepository());
            run.setPipelineId(pipeline.getId());
            run.setVersion(version);
            run.setRevisionName(gitManager.getRevisionName(version));
        }
        run.setStatus(TaskStatus.RUNNING);
        run.setCommitStatus(CommitStatus.NOT_COMMITTED);
        run.setLastChangeCommitTime(DateUtils.now());
        run.setPodId(getRootPodIDFromPipeline(run));
        Optional.ofNullable(parentRun).map(PipelineRun::getId).ifPresent(run::setParentRunId);
        run.convertParamsToString(configuration.getParameters());
        run.setTimeout(configuration.getTimeout());
        run.setDockerImage(configuration.getDockerImage());
        run.setActualDockerImage(Optional.ofNullable(tool).map(Tool::getImage).orElse(configuration.getDockerImage()));
        run.setPlatform(Optional.ofNullable(toolVersion).map(ToolVersion::getPlatform).orElse("linux"));
        instance.setNodePlatform(run.getPlatform());
        run.setCmdTemplate(determinateCmdTemplateForRun(configuration));
        run.setNodeCount(configuration.getNodeCount());
        setRunPrice(instance, run);
        run.setSshPassword(PasswordGenerator.generatePassword());
        run.setOwner(authManager.getAuthorizedUser());
        if (CollectionUtils.isNotEmpty(entityIds)) {
            run.setEntitiesIds(entityIds);
        }
        run.setSensitive(sensitive);
        run.setConfigurationId(configurationId);
        run.setExecutionPreferences(Optional.ofNullable(configuration.getExecutionPreferences())
                .orElse(ExecutionPreferences.getDefault()));
        if (StringUtils.isNotBlank(configuration.getPrettyUrl())) {
            validatePrettyUrlFree(configuration.getPrettyUrl());
            run.setPrettyUrl(configuration.getPrettyUrl());
        }
        if (instance.getSpot() != null &&!instance.getSpot()) {
            run.setNonPause(configuration.isNonPause());
        }
        return run;
    }

    private boolean checkRunForSensitivity(final Map<String, PipeConfValueVO> parameters) {
        List<Long> datastorageIds = MapUtils.emptyIfNull(parameters).entrySet().stream()
                .filter(v -> v.getKey().equals(CP_CAP_LIMIT_MOUNTS))
                .map(Map.Entry::getValue)
                .flatMap(pipeConfValueVO -> {
                            String limitMounts = pipeConfValueVO.getValue();
                            if (LIMIT_MOUNTS_NONE.equalsIgnoreCase(limitMounts)) {
                                return Stream.empty();
                            }
                            return Arrays.stream(commaDelimitedListToStringArray(limitMounts))
                                         .map(Long::valueOf);
                        }
                )
                .collect(Collectors.toList());
        if (datastorageIds.isEmpty()) {
            return false;
        }
        return dataStorageManager.getDatastoragesByIds(datastorageIds)
                .stream().anyMatch(AbstractDataStorage::isSensitive);
    }

    private String[] commaDelimitedListToStringArray(final String limitMounts) {
        return org.springframework.util.StringUtils.commaDelimitedListToStringArray(limitMounts);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun save(PipelineRun pipelineRun) {
        pipelineRunDao.createPipelineRun(pipelineRun);
        return pipelineRun;
    }

    public Tool getToolForRun(PipelineConfiguration configuration) {
        return toolManager.resolveSymlinks(configuration.getDockerImage());
    }

    /**
     * Updates pipeline run sids
     * @param runId {@link PipelineRun} id for pipeline run which run sids should be updated
     * @param runSids - a list of identities (user names or groups) that have access to run
     * @return updated pipeline {@link RunSid}
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateRunSids(Long runId, List<RunSid> runSids) {
        PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, runId));
        pipelineRunDao.deleteRunSids(runId);
        pipelineRunDao.createRunSids(runId, runSids);
        pipelineRun.setRunSids(runSids);
        return pipelineRun;
    }

    /**
     * Update whole pipeline info (use instead of calling few specific update methods if need).
     * @param run {@link PipelineRun} which will be updated
     * @return updated pipeline run
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateRunInfo(final PipelineRun run) {
        final Long runId = run.getId();
        final PipelineRun loadedRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(loadedRun,
                       messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, runId));
        pipelineRunDao.updateRun(run);
        return run;
    }

    /**
     * Updates run's tags
     * @param runId is ID of pipeline run which tags should be updated
     * @param newTags object, containing a map with tags to set for a pipeline
     * @param overwrite
     * @return
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateTags(final Long runId, final TagsVO newTags, final boolean overwrite) {
        final PipelineRun run = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(run,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, runId));
        if (overwrite) {
            run.setTags(newTags.getTags());
        } else {
            final Map<String, String> currentTags = new HashMap<>(MapUtils.emptyIfNull(run.getTags()));
            currentTags.putAll(MapUtils.emptyIfNull(newTags.getTags()));
            run.setTags(currentTags);
        }
        pipelineRunDao.updateRunTags(run);
        return run;
    }

    /**
     * Returns list of runs, that possibly produced spending during the given period
     *
     * @param start beginning of evaluating period
     * @param end ending of evaluating period
     * @return run with statuses adjusted
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public List<PipelineRun> loadRunsActivityStats(final LocalDateTime start, final LocalDateTime end) {
        final List<PipelineRun> runs = pipelineRunDao.loadPipelineRunsActiveInPeriod(start, end);
        final List<Long> runIds = runs.stream()
            .map(BaseEntity::getId)
            .collect(Collectors.toList());

        if (CollectionUtils.isEmpty(runIds)) {
            return Collections.emptyList();
        }

        final Map<Long, List<RunStatus>> runStatuses = runStatusManager.loadRunStatus(runIds);

        return runs.stream()
            .peek(run -> run.setRunStatuses(runStatuses.get(run.getId())))
            .collect(Collectors.toList());
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updateRunsTags(final Collection<PipelineRun> runs) {
        pipelineRunDao.updateRunsTags(runs);
    }

    /**
     * Updates run state reason message which was retrieved from instance {@StateReason}
     * @param run {@link PipelineRun} for pipeline run which state reason message should be updated
     * @param stateReasonMessage message that should be updated
     * @return
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateStateReasonMessage(PipelineRun run, String stateReasonMessage) {
        run.setStateReasonMessage(stateReasonMessage);
        pipelineRunDao.updateRun(run);
        return run;
    }

    /**
     * Updates run state reason message which was retrieved from instance
     * @param runId - the ID of pipeline run which state reason message should be updated
     * @param stateReasonMessage message that should be updated
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun updateStateReasonMessageById(final Long runId, final String stateReasonMessage) {
        final PipelineRun run = pipelineRunDao.loadPipelineRun(runId);
        run.setStateReasonMessage(stateReasonMessage);
        pipelineRunDao.updateRun(run);
        return run;
    }

    /**
     * Restarts spot run
     * @param run {@link PipelineRun} which will be restart
     * @return Restarted pipeline run
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun restartRun(final PipelineRun run) {
        final PipelineConfiguration configuration = configurationManager.getConfigurationFromRun(run);
        final PipelineRun restartedRun = createRestartRun(run);
        final Tool tool = getToolForRun(configuration);
        final List<String> endpoints = configuration.isEraseRunEndpoints() ?
                Collections.emptyList() : tool.getEndpoints();
        configuration.setSecretName(tool.getSecretName());
        configuration.setPodAssignPolicy(
            RunAssignPolicy.builder()
                .selector(
                    RunAssignPolicy.PodAssignSelector.builder()
                            .label(KubernetesConstants.RUN_ID_LABEL)
                            .value(restartedRun.getId().toString()).build())
                .build()
        );
        final String launchedCommand = pipelineLauncher.launch(restartedRun, configuration, endpoints, null);
        restartedRun.setActualCmd(launchedCommand);
        save(restartedRun);

        final RestartRun restartRun = new RestartRun();
        restartRun.setParentRunId(run.getId());
        restartRun.setRestartedRunId(restartedRun.getId());
        restartRun.setDate(DateUtils.now());
        restartRunManager.createRestartRun(restartRun);
        return restartedRun;
    }

    /**
     * Terminates paused run.
     *
     * It terminates the run cloud instance if it exists and stops the run.
     *
     * @param runId {@link PipelineRun} id for pipeline run.
     * @return Terminated pipeline run.
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun terminateRun(final Long runId) {
        final PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_RUN_PIPELINES_NOT_FOUND, runId));
        Assert.state(pipelineRun.getStatus() == TaskStatus.PAUSED,
                messageHelper.getMessage(MessageConstants.ERROR_RUN_TERMINATION_WRONG_STATUS, runId,
                        pipelineRun.getStatus()));
        pipelineRun.setStatus(TaskStatus.STOPPED);
        pipelineRun.setEndDate(DateUtils.now());
        runCRUDService.updateRunStatus(pipelineRun);
        nodesManager.terminateRun(pipelineRun);
        return pipelineRun;
    }

    /**
     * Creates and attaches new disk by the given request to the run.
     *
     * @param runId {@link PipelineRun} id for pipeline run.
     * @param request Attaching disk request.
     * @return Updated pipeline run.
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun attachDisk(final Long runId, final DiskAttachRequest request) {
        final PipelineRun pipelineRun = pipelineRunDao.loadPipelineRun(runId);
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_RUN_PIPELINES_NOT_FOUND, runId));
        Assert.state(pipelineRun.getStatus() == TaskStatus.RUNNING || pipelineRun.getStatus().isPause(),
                messageHelper.getMessage(MessageConstants.ERROR_RUN_DISK_ATTACHING_WRONG_STATUS, runId,
                        pipelineRun.getStatus()));
        Assert.notNull(request.getSize(),
                messageHelper.getMessage(MessageConstants.ERROR_RUN_DISK_SIZE_NOT_FOUND));
        Assert.isTrue(request.getSize() > 0,
                messageHelper.getMessage(MessageConstants.ERROR_INSTANCE_DISK_IS_INVALID, request.getSize()));
        nodesManager.attachDisk(pipelineRun, request);
        return pipelineRun;
    }

    /**
     * Generates launch command for 'pipe run' CLI method.
     * Note: runParameters should be at the end of the command.
     *
     * @param runVO {@link PipeRunCmdStartVO} contains input arguments for launch command
     * @return launch command
     */
    public String generateLaunchCommand(final PipeRunCmdStartVO runVO) {
        Assert.notNull(runVO.getPipelineStart(), "Pipeline start must be specified");
        return new PipeRunCmdBuilder(runVO)
                .name()
                .config()
                .parameters()
                .yes()
                .instanceDisk()
                .instanceType()
                .dockerImage()
                .cmdTemplate()
                .timeout()
                .quite()
                .instanceCount()
                .sync()
                .priceType()
                .regionId()
                .parentNode()
                .nonPause()
                .runParameters()
                .build();
    }

    @Transactional(propagation = Propagation.SUPPORTS)
    public List<PipelineRun> loadRunsByStatuses(final List<TaskStatus> statuses) {
        return pipelineRunDao.loadRunsByStatuses(statuses);
    }

    /**
     * Adjusts run price per hour including provided node disks.
     *
     * @param runId of {@link PipelineRun} to update price for.
     * @param disks of {@link PipelineRun} instance.
     * @return Updated pipeline run.
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public PipelineRun adjustRunPricePerHourToDisks(final Long runId, final List<InstanceDisk> disks) {
        final PipelineRun run = loadPipelineRun(runId, false);
        final RunInstance instance = run.getInstance();
        if (disks.isEmpty()) {
            LOGGER.warn("Run #{} price per hour won't be adjusted since no disks are provided.", runId);
            return run;
        }
        final BigDecimal pricePerHour = Optional.ofNullable(instance)
                .filter(runInstance -> !runInstance.isEmpty())
                .map(runInstance -> instanceOfferManager.getInstanceEstimatedPrice(runInstance.getNodeType(),
                        getTotalSize(disks), runInstance.getSpot(), runInstance.getCloudRegionId()))
                .map(InstancePrice::getPricePerHour)
                .map(this::scaledForUser)
                .orElse(run.getPricePerHour());
        run.setPricePerHour(pricePerHour);
        LOGGER.debug("Adjusted price per hour for run #{} to {}", runId, run.getPricePerHour());
        return updateRunInfo(run);
    }

    public Optional<PipelineRun> loadActiveRunsByPodIP(final String ip) {
        return pipelineRunDao.loadRunByPodIP(ip, Arrays.stream(TaskStatus.values())
                .filter(status -> !status.isFinal())
                .collect(Collectors.toList()));
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void stopServerlessRun(final Long runId) {
        stop(runId);
        stopServerlessRunManager.deleteByRunId(runId);
    }

    public List<PipelineRunWithTool> loadRunsWithTools(final List<Long> runIds) {
        final List<PipelineRun> runs = runCRUDService.loadRunsByIds(runIds);
        if (CollectionUtils.isEmpty(runs)) {
            return Collections.emptyList();
        }
        final Map<String, DockerRegistry> allRegistries = ListUtils.emptyIfNull(
                dockerRegistryManager.loadAllDockerRegistry()).stream()
                .collect(Collectors.toMap(DockerRegistry::getPath, Function.identity()));
        final Map<String, Set<String>> imagesByRegistries = runs.stream()
                .map(PipelineRun::getDockerImage)
                .map(this::parseDockerImage)
                .filter(Objects::nonNull)
                .collect(Collectors.groupingBy(Pair::getKey, mapping(Pair::getValue, toSet())));
        final Map<String, Tool> tools = imagesByRegistries.entrySet().stream()
                .flatMap(entry -> ListUtils.emptyIfNull(toolManager.loadAllByRegistryAndImageIn(
                        parseRegistryId(entry.getKey(), allRegistries), entry.getValue())).stream())
                .collect(Collectors.toMap(this::buildRegistryPath, Function.identity()));
        return runs.stream()
                .map(run -> PipelineRunWithTool.builder()
                        .pipelineRun(run)
                        .tool(tools.get(trimToolVersion(run.getDockerImage())))
                        .build())
                .collect(Collectors.toList());
    }

    public List<PipelineRun> loadRunsByPoolId(final Long poolId) {
        nodePoolManager.load(poolId);
        return runCRUDService.loadRunsByPoolId(poolId);
    }

    public List<RunInfo> loadRunsByParentId(final Long parentId) {
        return ListUtils.emptyIfNull(pipelineRunDao.loadRunsByParentRuns(Collections.singletonList(parentId))).stream()
                .map(run -> RunInfo.builder()
                        .runId(run.getId())
                        .status(run.getStatus())
                        .startDate(run.getStartDate())
                        .endDate(run.getEndDate())
                        .build())
                .collect(Collectors.toList());
    }

    private int getTotalSize(final List<InstanceDisk> disks) {
        return (int) disks.stream().mapToLong(InstanceDisk::getSize).sum();
    }

    private void adjustInstanceDisk(final PipelineConfiguration configuration) {
        long imageSizeBytes = toolManager.getCurrentImageSize(configuration.getDockerImage());
        long requiredDiskForImageBytes = imageSizeBytes
                * preferenceManager.getPreference(SystemPreferences.CLUSTER_DOCKER_EXTRA_MULTI)
                * preferenceManager.getPreference(SystemPreferences.CLUSTER_INSTANCE_HDD_EXTRA_MULTI);
        long requiredDiskSizeGb = (long)Math.ceil((double)requiredDiskForImageBytes / DIVIDER_TO_GB);
        long requestedDiskSize = NumberUtils.createLong(configuration.getInstanceDisk());
        LOGGER.debug("Retrieved image size: {} b, image size on disk: {} Gb, total disk size: {} Gb",
                imageSizeBytes, requiredDiskSizeGb, requiredDiskSizeGb + requestedDiskSize);
        configuration.setEffectiveDiskSize(Math.toIntExact(requiredDiskSizeGb + requestedDiskSize));
    }


    /**
     * @deprecated because its underlying method is deprecated.
     * @see PipelineRunDao#eagerSearchPipelineParentRuns(PagingRunFilterVO, PipelineRunFilterVO.ProjectFilter)
     */
    @Deprecated
    private PagedResult<List<PipelineRun>> eagerSearchParentRuns(PagingRunFilterVO filter,
                                                                 PipelineRunFilterVO.ProjectFilter projectFilter) {
        log.debug("Executing deprecated eager search parent runs request...");
        List<PipelineRun> runs = pipelineRunDao.eagerSearchPipelineParentRuns(filter, projectFilter);
        Integer totalCount = pipelineRunDao.countRootRuns(filter, projectFilter);
        return new PagedResult<>(runs, totalCount);
    }

    private PagedResult<List<PipelineRun>> searchParentRuns(PagingRunFilterVO filter,
                                                            PipelineRunFilterVO.ProjectFilter projectFilter) {
        List<PipelineRun> runs = pipelineRunDao.searchPipelineParentRuns(filter, projectFilter);
        Integer totalCount = pipelineRunDao.countRootRuns(filter, projectFilter);
        return new PagedResult<>(runs, totalCount);
    }

    private PagedResult<List<PipelineRun>> searchRuns(PagingRunFilterVO filter,
                                                      PipelineRunFilterVO.ProjectFilter projectFilter) {
        List<PipelineRun> runs = pipelineRunDao.searchPipelineRuns(filter, projectFilter);
        int totalCount = pipelineRunDao.countFilteredPipelineRuns(filter, projectFilter);
        return new PagedResult<>(runs, totalCount);
    }

    private void setParentInstance(PipelineRun run, Long parentNodeId) {
        PipelineRun parentRun = runCRUDService.loadRunById(parentNodeId);
        run.setInstance(parentRun.getInstance());
    }

    private void runClusterWorkers(PipelineRun run, PipelineStart runVO, String version, Pipeline pipeline,
                                   PipelineConfiguration configuration) {
        String parentId = Long.toString(run.getId());
        Integer nodeCount = configuration.getNodeCount();
        final PipelineConfiguration workerConfigurationTemplate = configurationManager.generateWorkerConfiguration(
                parentId, runVO, configuration, false, true);
        for (int i = 0; i < nodeCount; i++) {
            final PipelineConfiguration workerConfiguration = workerConfigurationTemplate.clone();
            launchPipeline(
                    workerConfiguration, pipeline, version, runVO.getInstanceType(), runVO.getConfigurationName(),
                    parentId, run.getId(), null, null, runVO.getRunSids(),
                    workerConfiguration.getNotifications()
            );
        }
    }

    private String determinateCmdTemplateForRun(PipelineConfiguration configuration) {
        if (StringUtils.isEmpty(configuration.getCmdTemplate())) {
            String defaultToolCommand = getToolForRun(configuration).getDefaultCommand();
            Assert.isTrue(!StringUtils.isEmpty(defaultToolCommand),
                    messageHelper.getMessage(MessageConstants.SETTING_IS_NOT_PROVIDED, "cmd_template"));
            return defaultToolCommand;
        } else {
            return configuration.getCmdTemplate();
        }
    }

    private void validateRunParameters(PipelineConfiguration configuration, Pipeline pipeline) {
        for (Map.Entry<String, PipeConfValueVO> param : configuration.getParameters().entrySet()) {
            boolean ifRequiredThenNotEmpty = !param.getValue().isRequired()
                    || !StringUtils.isEmpty(param.getValue().getValue());
            String pipelineName = pipeline == null ? "CMD" : pipeline.getName();
            Assert.isTrue(
                    ifRequiredThenNotEmpty,
                    messageHelper.getMessage(
                            MessageConstants.ERROR_RUN_PARAMETER_IS_REQUIRED,
                            param.getKey(),
                            pipelineName
                    )
            );
        }
    }

    private String getRootPodIDFromPipeline(PipelineRun run) {
        String runId = String.valueOf(run.getId());
        return normalizePodName(run.getPipelineName(), runId);
    }

    private String getRootPodIDFromTool(String image, Long id) {
        String runId = String.valueOf(id);
        int suffixLength = POD_ID_LENGTH - (image.length() + runId.length() + 1);
        String name = image;
        if (suffixLength > 1) {
            String suffix = PasswordGenerator.generateRandomString(suffixLength);
            name = String.format("%s-%s", image, suffix);
        }
        return normalizePodName(name, runId);
    }

    // PodId should contain only a-z 0-9 and -, it also should be smaller then 63 characters
    private String normalizePodName(String name, String runId) {
        String podName = name.trim().toLowerCase();
        podName = podName.replaceAll(KubernetesConstants.KUBE_NAME_REGEXP, "-");
        if (podName.length() + runId.length() + 1 > POD_ID_LENGTH) {
            podName = podName.substring(0, POD_ID_LENGTH - runId.length() - 1);
        }
        return String.format("%s-%s", podName, runId);
    }

    private RunInstance configureRunInstance(PipelineConfiguration configuration, AbstractCloudRegion region) {
        RunInstance instance = new RunInstance();
        instance.setNodeDisk(Optional.ofNullable(configuration.getInstanceDisk())
                .map(disk -> Integer.parseInt(configuration.getInstanceDisk()))
                .orElse(null));
        instance.setEffectiveNodeDisk(Optional.ofNullable(configuration.getEffectiveDiskSize())
                .orElse(instance.getNodeDisk()));
        instance.setNodeType(configuration.getInstanceType());
        instance.setNodeImage(configuration.getInstanceImage());
        Optional.ofNullable(region).map(AbstractCloudRegion::getId).ifPresent(instance::setCloudRegionId);
        Optional.ofNullable(region).map(AbstractCloudRegion::getProvider).ifPresent(instance::setCloudProvider);
        boolean defaultUseSpot = preferenceManager.getPreference(SystemPreferences.CLUSTER_SPOT);
        instance.setSpot(Optional.ofNullable(configuration.getIsSpot()).orElse(defaultUseSpot));
        return instance;
    }

    private PipelineRun updatePipelineStatus(Long id, TaskStatus status, PipelineRun pipelineRun) {
        Assert.notNull(pipelineRun,
                messageHelper.getMessage(MessageConstants.ERROR_PIPELINE_NOT_FOUND, id));
        Assert.notNull(status, messageHelper
                .getMessage(MessageConstants.ERROR_PARAMETER_REQUIRED, "status",
                        PipelineRun.class.getSimpleName()));
        Assert.isTrue(status != TaskStatus.RUNNING,
                messageHelper.getMessage(MessageConstants.ERROR_WRONG_RUN_STATUS_UPDATE, status));
        if (status.isFinal()) {
            pipelineRun.setEndDate(DateUtils.now());
        }
        pipelineRun.setStatus(status);
        pipelineRun.setTerminating(status.isFinal());

        dataStorageManager.analyzePipelineRunsParameters(Arrays.asList(pipelineRun));
        runCRUDService.updateRunStatus(pipelineRun);
        setParent(pipelineRun);
        return pipelineRun;
    }

    private void setParent(PipelineRun run) {
        if (run.getPipelineId() != null && run.getPipelineId() != 0L) {
            run.setParent(pipelineManager.load(run.getPipelineId()));
        }
    }

    private void checkCommitRunStatus(PipelineRun run) {
        Date currentTime = DateUtils.now();
        long secsFromLastChange = (currentTime.getTime() - run.getLastChangeCommitTime().getTime()) / MILLS_IN_SEC;
        int commitTimeout = preferenceManager.getPreference(SystemPreferences.COMMIT_TIMEOUT);
        if (run.getCommitStatus() == CommitStatus.COMMITTING && secsFromLastChange > commitTimeout) {
            LOGGER.warn(messageHelper.getMessage(
                    MessageConstants.WARN_COMMIT_TIMEOUT_HAS_EXPIRED,
                    run.getCommitStatus(),
                    secsFromLastChange,
                    CommitStatus.FAILURE));
            run.setCommitStatus(CommitStatus.FAILURE);
            run.setLastChangeCommitTime(currentTime);
            updatePipelineCommitStatus(run);
        }
    }

    public List<PipelineRunParameter> replaceParametersWithEnvVars(List<PipelineRunParameter> params,
                                                            Map<String, String> envVars) {
        if (CollectionUtils.isEmpty(params) || MapUtils.isEmpty(envVars)) {
            return params;
        }
        params.forEach(p -> p.setResolvedValue(p.getValue()));
        final List<PipelineRunParameter> paramsWithPossibleEnvVars = params.stream()
                .filter(p -> StringUtils.isNotBlank(p.getValue()) &&
                        p.getValue().contains("$"))
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(paramsWithPossibleEnvVars)) {
            return params;
        }
        envVars.forEach((key, value) -> {
            Pattern pattern = Pattern.compile("[$]" + key + "[^a-zA-Z0-9_]");
            Pattern patternWithBraces = Pattern.compile("[$]" + key + "$|[$][{]" + key + "[}]");
            paramsWithPossibleEnvVars.forEach(param ->
                    param.setResolvedValue(
                        replaceEnvVarInParameter(pattern, patternWithBraces, key, value, param.getResolvedValue())));
        });
        return params;
    }

    private String replaceEnvVarInParameter(final Pattern pattern,
                                            final Pattern patternWithBraces,
                                            final String envVarName,
                                            final String envVarValue,
                                            final String parameter) {
        if (StringUtils.isBlank(parameter) || StringUtils.isBlank(envVarName)
                || StringUtils.isBlank(envVarValue)) {
            return parameter;
        }
        try {
            String resolvedParameter = parameter;
            Matcher matcher = pattern.matcher(resolvedParameter);
            while (matcher.find()) {
                char lastSymbol = matcher.group().toCharArray()[matcher.group().length() - 1];
                resolvedParameter = matcher.replaceAll(envVarValue + lastSymbol);
            }
            return patternWithBraces.matcher(resolvedParameter).replaceAll(envVarValue);
        } catch (IllegalArgumentException e) {
            LOGGER.trace(e.getMessage(), e);
            return parameter;
        }
    }

    PipelineRunFilterVO.ProjectFilter resolveProjectFiltering(PipelineRunFilterVO filter) {
        if (CollectionUtils.isEmpty(filter.getProjectIds())) {
            return null;
        }
        List<Long> pipelineIds = new ArrayList<>();
        List<Long> configurationIds = new ArrayList<>();
        filter.getProjectIds().stream()
                .map(id -> folderApiService.load(id))
                .forEach(folder -> collectChildren(folder, pipelineIds, configurationIds));
        return new PipelineRunFilterVO.ProjectFilter(pipelineIds, configurationIds);
    }

    private void collectChildren(Folder folder, List<Long> pipelineIds, List<Long> configurationIds) {
        Map<AclClass, List<Long>> projectChildren = folder.collectChildren();
        if (projectChildren.containsKey(AclClass.PIPELINE)) {
            pipelineIds.addAll(projectChildren.get(AclClass.PIPELINE));
        }
        if (projectChildren.containsKey(AclClass.CONFIGURATION)) {
            configurationIds.addAll(projectChildren.get(AclClass.CONFIGURATION));
        }
    }

    private void validatePrettyUrlFree(String url) {
        pipelineRunDao.loadRunByPrettyUrl(url).ifPresent(r -> {
            throw new IllegalArgumentException(
                    messageHelper.getMessage(MessageConstants.ERROR_RUN_PRETTY_URL_IN_USE, url, r.getId()));
        });
    }

    private void setRunPrice(final RunInstance instance, final PipelineRun run) {
        if (!instance.isEmpty()) {
            run.setInstance(instance);
            InstancePrice runPrice = instanceOfferManager.getInstanceEstimatedPrice(
                    instance.getNodeType(), instance.getEffectiveNodeDisk(), instance.getSpot(),
                    instance.getCloudRegionId());
            LOGGER.debug("Expected price per hour: {}", runPrice.getPricePerHour());
            run.setPricePerHour(scaledForUser(runPrice.getPricePerHour()));
            run.setComputePricePerHour(scaledForBilling(runPrice.getComputePricePerHour()));
            run.setDiskPricePerHour(scaledForBilling(runPrice.getDiskPricePerHour()));
        }
    }

    private BigDecimal scaledForUser(final double price) {
        return scaled(price, USER_PRICE_SCALE);
    }

    private BigDecimal scaledForBilling(final double price) {
        return scaled(price, BILLING_PRICE_SCALE);
    }

    private BigDecimal scaled(final double price, final int scale) {
        return BigDecimal.valueOf(price).setScale(scale, RoundingMode.HALF_EVEN);
    }

    private PipelineRun createRestartRun(final PipelineRun run) {
        PipelineRun restartedRun = new PipelineRun();
        Long runId = pipelineRunDao.createRunId();
        restartedRun.setId(runId);
        restartedRun.setStartDate(DateUtils.now());
        restartedRun.setProlongedAtTime(DateUtils.nowUTC());

        Optional<Pipeline> pipeline = Optional.ofNullable(run.getPipelineId())
                .map(pipelineId -> pipelineManager.load(pipelineId));

        pipeline.ifPresent(p -> {
            restartedRun.setPipelineName(p.getName());
            restartedRun.setRepository(p.getRepository());
            restartedRun.setPipelineId(p.getId());
            restartedRun.setVersion(run.getVersion());
            restartedRun.setRevisionName(gitManager.getRevisionName(run.getVersion()));
        });

        if (!pipeline.isPresent()) {
            fillMissingPipelineFields(restartedRun);
        }

        restartedRun.setStatus(TaskStatus.RUNNING);
        restartedRun.setCommitStatus(CommitStatus.NOT_COMMITTED);
        restartedRun.setLastChangeCommitTime(DateUtils.now());
        restartedRun.setPodId(getRootPodIDFromPipeline(restartedRun));
        restartedRun.setParams(run.getParams());
        restartedRun.parseParameters();
        restartedRun.setTimeout(run.getTimeout());
        restartedRun.setDockerImage(run.getDockerImage());
        restartedRun.setActualDockerImage(run.getActualDockerImage());
        restartedRun.setPlatform(run.getPlatform());
        restartedRun.setCmdTemplate(run.getCmdTemplate());
        restartedRun.setNodeCount(run.getNodeCount());
        RunInstance instance = copyInstance(run.getInstance());
        restartedRun.setInstance(instance);
        setRunPrice(instance, restartedRun);
        restartedRun.setSshPassword(PasswordGenerator.generatePassword());
        restartedRun.setOwner(run.getOwner());
        restartedRun.setEntitiesIds(run.getEntitiesIds());
        restartedRun.setSensitive(run.getSensitive());
        restartedRun.setConfigurationId(run.getConfigurationId());
        restartedRun.setExecutionPreferences(run.getExecutionPreferences());
        restartedRun.setRunSids(run.getRunSids());
        restartedRun.setPrettyUrl(run.getPrettyUrl());
        restartedRun.setNonPause(run.isNonPause());
        return restartedRun;
    }

    private void fillMissingPipelineFields(final PipelineRun restartedRun) {
        restartedRun.setPipelineName(PipelineRun.DEFAULT_PIPELINE_NAME);
        restartedRun.setRepository(EMPTY_STRING);
        restartedRun.setPipelineId(EMPTY_PIPELINE_ID);
        restartedRun.setVersion(EMPTY_STRING);
        restartedRun.setRevisionName(EMPTY_STRING);
    }

    private RunInstance copyInstance(final RunInstance instance) {
        return Optional.ofNullable(instance).map(i -> {
            RunInstance runInstance = new RunInstance();
            runInstance.setCloudRegionId(i.getCloudRegionId());
            runInstance.setNodeDisk(i.getNodeDisk());
            runInstance.setEffectiveNodeDisk(i.getEffectiveNodeDisk());
            runInstance.setNodeImage(i.getNodeImage());
            runInstance.setNodeType(i.getNodeType());
            runInstance.setSpot(i.getSpot());
            runInstance.setCloudProvider(i.getCloudProvider());
            runInstance.setNodePlatform(i.getNodePlatform());
            return runInstance;
        }).orElse(new RunInstance());
    }

    private Pair<String, String> parseDockerImage(final String dockerImage) {
        final Matcher matcher = REPOSITORY_AND_IMAGE.matcher(dockerImage);

        if (!matcher.find()) {
            LOGGER.error("Failed to parse docker image ['{}']", dockerImage);
            return null;
        }

        final String repository = matcher.group(1);
        final String imageWithTag = matcher.group(2);
        final String imageName = getImageWithoutTag(imageWithTag);

        return Pair.of(repository, imageName);
    }

    private Long parseRegistryId(final String registry, final Map<String, DockerRegistry> allRegistries) {
        return allRegistries.getOrDefault(registry, new DockerRegistry()).getId();
    }

    private String buildRegistryPath(final Tool tool) {
        return formatRegistryPath(tool.getRegistry(), tool.getImage());
    }

    private String formatRegistryPath(final String registry, final String image) {
        return String.format("%s/%s", registry, image);
    }

    private String trimToolVersion(final String docketImage) {
        final Pair<String, String> parsedImage = parseDockerImage(docketImage);
        return Objects.isNull(parsedImage)
                ? null
                : formatRegistryPath(parsedImage.getKey(), parsedImage.getValue());
    }

    private void checkRunLaunchLimits(final PipelineStart runVO) {
        final int totalStaticNodesCount = Optional.of(runVO)
                                              .map(PipelineStart::getNodeCount)
                                              .orElse(0) + 1;
        runLimitsService.checkRunLaunchLimits(totalStaticNodesCount);
    }

    private void updateMetadataEntities(final PipelineRun run) {
        final List<Long> entitiesIds = run.getEntitiesIds();
        if (CollectionUtils.isEmpty(entitiesIds)) {
            return;
        }
        final Optional<PipelineRunParameter> runStatusParameter = ListUtils
                .emptyIfNull(run.getPipelineRunParameters()).stream()
                .filter(runParameter -> CP_REPORT_RUN_STATUS.equals(runParameter.getName()))
                .findFirst();
        if (!runStatusParameter.isPresent()) {
            return;
        }
        final String dataKey = runStatusParameter.get().getValue();
        if (StringUtils.isBlank(dataKey)) {
            LOGGER.error("Parameter {} was specified for pipeline run {} but empty", CP_REPORT_RUN_STATUS, run.getId());
            return;
        }
        final RunStatusMetadata runStatusMetadata = RunStatusMetadata.builder()
                .runId(run.getId())
                .status(run.getStatus().name())
                .message(run.getStateReasonMessage())
                .startDate(run.getStartDate())
                .endDate(run.getEndDate())
                .build();

        final Optional<String> reportEndDateParameter = run.getParameterValue(CP_REPORT_RUN_PROCESSED_DATE);

        final List<MetadataEntity> metadataEntities = metadataEntityManager
                .loadEntitiesByIds(new HashSet<>(entitiesIds)).stream()
                .peek(metadataEntity -> {
                    if (Objects.isNull(metadataEntity.getData())) {
                        metadataEntity.setData(new HashMap<>());
                    }
                    addRunStatusMetadata(metadataEntity.getData(), dataKey, runStatusMetadata);
                    if (run.getStatus().isFinal() && Objects.nonNull(run.getEndDate()) &&
                            reportEndDateParameter.isPresent()) {
                        addProcessedDate(metadataEntity.getData(), reportEndDateParameter.get(),
                                run.getEndDate());
                    }
                })
                .collect(Collectors.toList());
        metadataEntityManager.updateMetadataEntities(metadataEntities);
    }

    private void addProcessedDate(final Map<String, PipeConfValue> data,
                                  final String key,
                                  final Date endDate) {
        final PipeConfValue value = new PipeConfValue(PipeConfValueType.DATE.toString(),
                DateUtils.formatDate(endDate));
        data.put(key, value);
    }

    private void addRunStatusMetadata(final Map<String, PipeConfValue> currentData,
                                      final String dataKey,
                                      final RunStatusMetadata runStatusMetadata) {
        final List<RunStatusMetadata> runStatuses = currentData.containsKey(dataKey)
                ? prepareExistingRunStatusMetadata(currentData.get(dataKey), runStatusMetadata)
                : Collections.singletonList(runStatusMetadata);
        final PipeConfValue value = new PipeConfValue(PipeConfValueType.JSON.toString(),
                JsonMapper.convertDataToJsonStringForQuery(runStatuses));
        currentData.put(dataKey, value);
    }

    private List<RunStatusMetadata> prepareExistingRunStatusMetadata(final PipeConfValue currentMetadata,
                                                                     final RunStatusMetadata runStatusMetadata) {
        if (Objects.isNull(currentMetadata)) {
            return Collections.emptyList();
        }
        final List<RunStatusMetadata> runStatuses = JsonMapper.parseData(currentMetadata.getValue(),
                new TypeReference<List<RunStatusMetadata>>() {});
        final Map<Long, RunStatusMetadata> statusesByRunId = ListUtils.emptyIfNull(runStatuses).stream()
                .collect(Collectors.toMap(RunStatusMetadata::getRunId, Function.identity()));
        statusesByRunId.put(runStatusMetadata.getRunId(), runStatusMetadata);
        return new ArrayList<>(statusesByRunId.values());
    }

    private void addRegionsToRestartedRuns(final List<RestartRun> restartedRuns) {
        final Map<Long, PipelineRun> runsById = ListUtils.emptyIfNull(
                pipelineRunDao.loadRunByIdIn(restartedRuns.stream()
                        .flatMap(restartRun -> Stream.of(restartRun.getRestartedRunId(), restartRun.getParentRunId()))
                        .distinct()
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList()))).stream()
                .collect(Collectors.toMap(PipelineRun::getId, Function.identity()));
        restartedRuns.forEach(restartRun -> {
            restartRun.setParentRunRegionId(
                    findRegionIdFromPipelineRun(runsById.get(restartRun.getParentRunId())));
            restartRun.setRestartedRunRegionId(
                    findRegionIdFromPipelineRun(runsById.get(restartRun.getRestartedRunId())));
        });
    }

    private Long findRegionIdFromPipelineRun(final PipelineRun run) {
        return Optional.ofNullable(run)
                .flatMap(pipelineRun -> Optional.ofNullable(pipelineRun.getInstance()))
                .map(RunInstance::getCloudRegionId)
                .orElse(null);

    }
}
