/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.ambari.server.controller.internal;

import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.PACKAGE_LIST;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariManagementControllerImpl;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.controller.ServiceComponentHostRequest;
import org.apache.ambari.server.controller.ServiceComponentHostResponse;
import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
import org.apache.ambari.server.controller.spi.NoSuchResourceException;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.RequestStatus;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.security.authorization.RoleAuthorization;
import org.apache.ambari.server.state.ClientConfigFileDefinition;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.PropertyInfo.PropertyType;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.ServiceOsSpecific;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.utils.SecretReference;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.AssistedInject;

/**
 * Resource provider for client config resources.
 */
public class ClientConfigResourceProvider extends AbstractControllerResourceProvider {


  // ----- Property ID constants ---------------------------------------------

  protected static final String COMPONENT_CLUSTER_NAME_PROPERTY_ID = "ServiceComponentInfo/cluster_name";
  protected static final String COMPONENT_SERVICE_NAME_PROPERTY_ID = "ServiceComponentInfo/service_name";
  protected static final String COMPONENT_COMPONENT_NAME_PROPERTY_ID = "ServiceComponentInfo/component_name";
  protected static final String HOST_COMPONENT_HOST_NAME_PROPERTY_ID =
          PropertyHelper.getPropertyId("HostRoles", "host_name");

  private final Gson gson;

  /**
   * The key property ids for a ClientConfig resource.
   */
  private static final Map<Resource.Type, String> keyPropertyIds = ImmutableMap.<Resource.Type, String>builder()
      .put(Resource.Type.Cluster, COMPONENT_CLUSTER_NAME_PROPERTY_ID)
      .put(Resource.Type.Service, COMPONENT_SERVICE_NAME_PROPERTY_ID)
      .put(Resource.Type.Component, COMPONENT_COMPONENT_NAME_PROPERTY_ID)
      .put(Resource.Type.Host, HOST_COMPONENT_HOST_NAME_PROPERTY_ID)
      .build();

  /**
   * The property ids for a ClientConfig resource.
   */
  private static final Set<String> propertyIds = Sets.newHashSet(
      COMPONENT_CLUSTER_NAME_PROPERTY_ID,
      COMPONENT_SERVICE_NAME_PROPERTY_ID,
      COMPONENT_COMPONENT_NAME_PROPERTY_ID,
      HOST_COMPONENT_HOST_NAME_PROPERTY_ID);

  private MaintenanceStateHelper maintenanceStateHelper;
  private static final Logger LOG = LoggerFactory.getLogger(ClientConfigResourceProvider.class);

  // ----- Constructors ----------------------------------------------------

  /**
   * Create a  new resource provider for the given management controller.
   *
   * @param managementController the management controller
   */
  @AssistedInject
  ClientConfigResourceProvider(@Assisted AmbariManagementController managementController) {
    super(Resource.Type.ClientConfig, propertyIds, keyPropertyIds, managementController);
    gson = new Gson();

    setRequiredGetAuthorizations(EnumSet.of(RoleAuthorization.HOST_VIEW_CONFIGS, RoleAuthorization.SERVICE_VIEW_CONFIGS, RoleAuthorization.CLUSTER_VIEW_CONFIGS));
  }

  // ----- ResourceProvider ------------------------------------------------

  @Override
  public RequestStatus createResources(Request request)
          throws SystemException,
          UnsupportedPropertyException,
          ResourceAlreadyExistsException,
          NoSuchParentResourceException {

    throw new SystemException("The request is not supported");
  }

  @Override
  public Set<Resource> getResourcesAuthorized(Request request, Predicate predicate)
          throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {

    Set<Resource> resources = new HashSet<>();

    final Set<ServiceComponentHostRequest> requests = new HashSet<>();

    for (Map<String, Object> propertyMap : getPropertyMaps(predicate)) {
      requests.add(getRequest(propertyMap));
    }

    Set<ServiceComponentHostResponse> responses = null;
    try {
      responses = getResources(new Command<Set<ServiceComponentHostResponse>>() {
        @Override
        public Set<ServiceComponentHostResponse> invoke() throws AmbariException {
          return getManagementController().getHostComponents(requests);
        }
      });
    } catch (Exception e) {
      throw new SystemException("Failed to get components ", e);
    }

    Map<String,ServiceComponentHostResponse> componentMap = new HashMap<>();

    // reduce set of sch responses to one sch response for every service component
    for (ServiceComponentHostResponse resp: responses) {
      String componentName = resp.getComponentName();
      if (!componentMap.containsKey(componentName)) {
        componentMap.put(resp.getComponentName(),resp);
      }
    }

    ServiceComponentHostRequest schRequest =  requests.iterator().next();
    String requestComponentName = schRequest.getComponentName();
    String requestServiceName = schRequest.getServiceName();
    String requestHostName =  schRequest.getHostname();

    Map<String,List<ServiceComponentHostResponse>> serviceToComponentMap = new HashMap<>();

    // sch response for the service components that have configFiles defined in the stack definition of the service
    List <ServiceComponentHostResponse> schWithConfigFiles = new ArrayList<>();

    Configuration configs = new Configuration();
    Map<String, String> configMap = configs.getConfigsMap();
    String TMP_PATH = configMap.get(Configuration.SERVER_TMP_DIR.getKey());
    String pythonCmd = configMap.get(Configuration.AMBARI_PYTHON_WRAP.getKey());
    List<String> pythonCompressFilesCmds = new ArrayList<>();
    List<File> commandFiles = new ArrayList<>();

    for (ServiceComponentHostResponse response : componentMap.values()){

      AmbariManagementController managementController = getManagementController();
      ConfigHelper configHelper = managementController.getConfigHelper();
      Cluster cluster = null;
      Clusters clusters = managementController.getClusters();
      try {
        cluster = clusters.getCluster(response.getClusterName());

        String serviceName = response.getServiceName();
        String componentName = response.getComponentName();
        String hostName = response.getHostname();
        String publicHostName = response.getPublicHostname();
        ComponentInfo componentInfo = null;
        String packageFolder = null;

        Service service = cluster.getService(serviceName);
        ServiceComponent component = service.getServiceComponent(componentName);
        StackId stackId = component.getDesiredStackId();

        componentInfo = managementController.getAmbariMetaInfo().
          getComponent(stackId.getStackName(), stackId.getStackVersion(), serviceName, componentName);

        packageFolder = managementController.getAmbariMetaInfo().
          getService(stackId.getStackName(), stackId.getStackVersion(), serviceName).getServicePackageFolder();

        String commandScript = componentInfo.getCommandScript().getScript();
        List<ClientConfigFileDefinition> clientConfigFiles = componentInfo.getClientConfigFiles();

        if (clientConfigFiles == null) {
          if (componentMap.size() == 1) {
            throw new SystemException("No configuration files defined for the component " + componentInfo.getName());
          } else {
            LOG.debug("No configuration files defined for the component {}", componentInfo.getName());
            continue;
          }
        }

        // service component hosts that have configFiles defined in the stack definition of the service
        schWithConfigFiles.add(response);

        if (serviceToComponentMap.containsKey(response.getServiceName())) {
          List <ServiceComponentHostResponse> schResponseList =  serviceToComponentMap.get(serviceName);
          schResponseList.add(response);
        } else {
          List <ServiceComponentHostResponse> schResponseList = new ArrayList<>();
          schResponseList.add(response);
          serviceToComponentMap.put(serviceName,schResponseList);
        }

        String resourceDirPath = configs.getResourceDirPath();
        String packageFolderAbsolute = resourceDirPath + File.separator + packageFolder;

        String commandScriptAbsolute = packageFolderAbsolute + File.separator + commandScript;


        Map<String, Map<String, String>> configurations = new TreeMap<>();
        Map<String, Long> configVersions = new TreeMap<>();
        Map<String, Map<PropertyType, Set<String>>> configPropertiesTypes = new TreeMap<>();
        Map<String, Map<String, Map<String, String>>> configurationAttributes = new TreeMap<>();

        Map<String, DesiredConfig> desiredClusterConfigs = cluster.getDesiredConfigs();

        //Get configurations and configuration attributes
        for (Map.Entry<String, DesiredConfig> desiredConfigEntry : desiredClusterConfigs.entrySet()) {

          String configType = desiredConfigEntry.getKey();
          DesiredConfig desiredConfig = desiredConfigEntry.getValue();
          Config clusterConfig = cluster.getConfig(configType, desiredConfig.getTag());

          if (clusterConfig != null) {
            Map<String, String> props = new HashMap<>(clusterConfig.getProperties());

            // Apply global properties for this host from all config groups
            Map<String, Map<String, String>> allConfigTags = null;
            allConfigTags = configHelper
              .getEffectiveDesiredTags(cluster, schRequest.getHostname());

            Map<String, Map<String, String>> configTags = new HashMap<>();

            for (Map.Entry<String, Map<String, String>> entry : allConfigTags.entrySet()) {
              if (entry.getKey().equals(clusterConfig.getType())) {
                configTags.put(clusterConfig.getType(), entry.getValue());
              }
            }

            Map<String, Map<String, String>> properties = configHelper
              .getEffectiveConfigProperties(cluster, configTags);

          if (!properties.isEmpty()) {
            for (Map<String, String> propertyMap : properties.values()) {
              props.putAll(propertyMap);
            }
          }

            configurations.put(clusterConfig.getType(), props);
            configVersions.put(clusterConfig.getType(), clusterConfig.getVersion());
            configPropertiesTypes.put(clusterConfig.getType(), clusterConfig.getPropertiesTypes());

            Map<String, Map<String, String>> attrs = new TreeMap<>();
            configHelper.cloneAttributesMap(clusterConfig.getPropertiesAttributes(), attrs);

            Map<String, Map<String, Map<String, String>>> attributes = configHelper
              .getEffectiveConfigAttributes(cluster, configTags);
            for (Map<String, Map<String, String>> attributesMap : attributes.values()) {
              configHelper.cloneAttributesMap(attributesMap, attrs);
            }
            configurationAttributes.put(clusterConfig.getType(), attrs);
          }
        }

        ConfigHelper.processHiddenAttribute(configurations, configurationAttributes, componentName, true);

        for (Map.Entry<String, Map<String, Map<String, String>>> configurationAttributesEntry : configurationAttributes.entrySet()) {
          Map<String, Map<String, String>> attrs = configurationAttributesEntry.getValue();
          // remove internal attributes like "hidden"
          attrs.remove("hidden");
        }

        // replace passwords on password references
        for (Map.Entry<String, Map<String, String>> configEntry : configurations.entrySet()) {
          String configType = configEntry.getKey();
          Map<String, String> configProperties = configEntry.getValue();
          Long configVersion = configVersions.get(configType);
          Map<PropertyType, Set<String>> propertiesTypes = configPropertiesTypes.get(configType);
          SecretReference.replacePasswordsWithReferences(propertiesTypes, configProperties, configType, configVersion);
        }

        Map<String, Set<String>> clusterHostInfo = null;
        ServiceInfo serviceInfo = null;
        String osFamily = null;
        clusterHostInfo = StageUtils.getClusterHostInfo(cluster);
        serviceInfo = managementController.getAmbariMetaInfo().getService(stackId.getStackName(),
          stackId.getStackVersion(), serviceName);
        try {
          clusterHostInfo = StageUtils.substituteHostIndexes(clusterHostInfo);
        } catch (AmbariException e) {
          // Before moving substituteHostIndexes to StageUtils, a SystemException was thrown in the
          // event an index could not be mapped to a host.  After the move, this was changed to an
          // AmbariException for consistency in the StageUtils class. To keep this method consistent
          // with how it behaved in the past, if an AmbariException is thrown, it is caught and
          // translated to a SystemException.
          throw new SystemException(e.getMessage(), e);
        }
        osFamily = clusters.getHost(hostName).getOsFamily();

        // Write down os specific info for the service
        ServiceOsSpecific anyOs = null;
        if (serviceInfo.getOsSpecifics().containsKey(AmbariMetaInfo.ANY_OS)) {
          anyOs = serviceInfo.getOsSpecifics().get(AmbariMetaInfo.ANY_OS);
        }

        ServiceOsSpecific hostOs = populateServicePackagesInfo(serviceInfo, osFamily);

        // Build package list that is relevant for host
        List<ServiceOsSpecific.Package> packages =
          new ArrayList<>();
        if (anyOs != null) {
          packages.addAll(anyOs.getPackages());
        }

        if (hostOs != null) {
          packages.addAll(hostOs.getPackages());
        }
        String packageList = gson.toJson(packages);

        String jsonConfigurations = null;
        Map<String, Object> commandParams = new HashMap<>();
        List<Map<String, String>> xmlConfigs = new LinkedList<>();
        List<Map<String, String>> envConfigs = new LinkedList<>();
        List<Map<String, String>> propertiesConfigs = new LinkedList<>();

        //Fill file-dictionary configs from metainfo
        for (ClientConfigFileDefinition clientConfigFile : clientConfigFiles) {
          Map<String, String> fileDict = new HashMap<>();
          fileDict.put(clientConfigFile.getFileName(), clientConfigFile.getDictionaryName());
          if (clientConfigFile.getType().equals("xml")) {
            xmlConfigs.add(fileDict);
          } else if (clientConfigFile.getType().equals("env")) {
            envConfigs.add(fileDict);
          } else if (clientConfigFile.getType().equals("properties")) {
            propertiesConfigs.add(fileDict);
          }
        }

        TreeMap<String, String> clusterLevelParams = null;
        TreeMap<String, String> ambariLevelParams = null;
        TreeMap<String, String> topologyCommandParams = new TreeMap<>();
        if (getManagementController() instanceof AmbariManagementControllerImpl){
          AmbariManagementControllerImpl controller = ((AmbariManagementControllerImpl)getManagementController());
          clusterLevelParams = controller.getMetadataClusterLevelParams(cluster, stackId);
          ambariLevelParams = controller.getMetadataAmbariLevelParams();

          Service s = cluster.getService(serviceName);
          ServiceComponent sc = s.getServiceComponent(componentName);
          ServiceComponentHost sch = sc.getServiceComponentHost(response.getHostname());

          topologyCommandParams = controller.getTopologyCommandParams(cluster.getClusterId(), serviceName, componentName, sch);
        }
        TreeMap<String, String> agentLevelParams = new TreeMap<>();
        agentLevelParams.put("hostname", hostName);
        agentLevelParams.put("public_hostname", publicHostName);

        commandParams.put(PACKAGE_LIST, packageList);
        commandParams.put("xml_configs_list", xmlConfigs);
        commandParams.put("env_configs_list", envConfigs);
        commandParams.put("properties_configs_list", propertiesConfigs);
        commandParams.put("output_file", componentName + "-configs" + Configuration.DEF_ARCHIVE_EXTENSION);
        commandParams.putAll(topologyCommandParams);

        Map<String, Object> jsonContent = new TreeMap<>();
        jsonContent.put("configurations", configurations);
        jsonContent.put("configurationAttributes", configurationAttributes);
        jsonContent.put("commandParams", commandParams);
        jsonContent.put("clusterHostInfo", clusterHostInfo);
        jsonContent.put("ambariLevelParams", ambariLevelParams);
        jsonContent.put("clusterLevelParams", clusterLevelParams);
        jsonContent.put("agentLevelParams", agentLevelParams);
        jsonContent.put("hostname", hostName);
        jsonContent.put("public_hostname", publicHostName);
        jsonContent.put("clusterName", cluster.getClusterName());
        jsonContent.put("serviceName", serviceName);
        jsonContent.put("role", componentName);
        jsonContent.put("componentVersionMap", cluster.getComponentVersionMap());

        jsonConfigurations = gson.toJson(jsonContent);

        File tmpDirectory = new File(TMP_PATH);
        if (!tmpDirectory.exists()) {
          try {
            tmpDirectory.mkdirs();
            tmpDirectory.setWritable(true, true);
            tmpDirectory.setReadable(true, true);
          } catch (SecurityException se) {
            throw new SystemException("Failed to get temporary directory to store configurations", se);
          }
        }
        File jsonFile = File.createTempFile(componentName, "-configuration.json", tmpDirectory);
        try {
          jsonFile.setWritable(true, true);
          jsonFile.setReadable(true, true);
        } catch (SecurityException e) {
          throw new SystemException("Failed to set permission", e);
        }

        PrintWriter printWriter = null;
        try {
          printWriter = new PrintWriter(jsonFile.getAbsolutePath());
          printWriter.print(jsonConfigurations);
          printWriter.close();
        } catch (FileNotFoundException e) {
          throw new SystemException("Failed to write configurations to json file ", e);
        }

        String cmd = pythonCmd + " " + commandScriptAbsolute + " generate_configs " + jsonFile.getAbsolutePath() + " " +
          packageFolderAbsolute + " " + TMP_PATH + File.separator + "structured-out.json" + " INFO " + TMP_PATH;

        commandFiles.add(jsonFile);
        pythonCompressFilesCmds.add(cmd);

      } catch (IOException e) {
        throw new SystemException("Controller error ", e);
      }
    }

    if (schWithConfigFiles.isEmpty()) {
      throw new SystemException("No configuration files defined for any component" );
    }

    Integer totalCommands = pythonCompressFilesCmds.size() * 2;
    Integer threadPoolSize = Math.min(totalCommands,configs.getExternalScriptThreadPoolSize());
    ExecutorService processExecutor = Executors.newFixedThreadPool(threadPoolSize);

    // put all threads that starts process to compress each component config files in the executor
    try {
      List<CommandLineThreadWrapper> pythonCmdThreads = executeCommands(processExecutor, pythonCompressFilesCmds);

      // wait for all threads to finish
      Integer timeout = configs.getExternalScriptTimeout();
      waitForAllThreadsToJoin(processExecutor, pythonCmdThreads, timeout);
    } finally {
      for (File each : commandFiles) {
        each.delete();
      }
    }

    if (StringUtils.isEmpty(requestComponentName)) {
      TarUtils tarUtils;
      String fileName;
      List <ServiceComponentHostResponse> schToTarConfigFiles = schWithConfigFiles;
      if (StringUtils.isNotEmpty(requestHostName)) {
        fileName = requestHostName + "(" + Resource.InternalType.Host.toString().toUpperCase()+")";
      } else if (StringUtils.isNotEmpty(requestServiceName)) {
        fileName = requestServiceName + "(" + Resource.InternalType.Service.toString().toUpperCase()+")";
        schToTarConfigFiles = serviceToComponentMap.get(requestServiceName);
      } else {
        fileName = schRequest.getClusterName() + "(" + Resource.InternalType.Cluster.toString().toUpperCase()+")";
      }
      tarUtils = new TarUtils(TMP_PATH, fileName, schToTarConfigFiles);
      tarUtils.tarConfigFiles();
    }

    Resource resource = new ResourceImpl(Resource.Type.ClientConfig);
    resources.add(resource);
    return resources;
  }

  /**
   *  Execute all external script commands
   * @param processExecutor {@link ExecutorService} executes the process when threads are available in the pool
   * @param commandLines List of {String} commands that starts the python process to compress config files
   * @return {@link CommandLineThreadWrapper}
   * @throws SystemException
   */
  private List<CommandLineThreadWrapper> executeCommands(final ExecutorService processExecutor, List<String> commandLines)
    throws SystemException {
    List <CommandLineThreadWrapper> commandLineThreadWrappers = new ArrayList<>();
    try {
      for (String commandLine : commandLines) {
        CommandLineThreadWrapper commandLineThreadWrapper = executeCommand(processExecutor,commandLine);
        commandLineThreadWrappers.add(commandLineThreadWrapper);
      }
    } catch (IOException e) {
      LOG.error("Failed to run generate client configs script for components");
      processExecutor.shutdownNow();
      throw new SystemException("Failed to run generate client configs script for components");
    }
    return  commandLineThreadWrappers;
  }


  /**
   * Execute external script command
   * @param processExecutor {@link ExecutorService} executes the process when threads are available in the pool
   * @param commandLine {String} command that starts the python process to compress config files
   * @return {@link CommandLineThreadWrapper}
   * @throws IOException
   */
  private CommandLineThreadWrapper executeCommand(final ExecutorService processExecutor, final String commandLine)
    throws IOException {
    ProcessBuilder builder = new ProcessBuilder(Arrays.asList(commandLine.split("\\s+")));
    builder.redirectErrorStream(true);
    Process process = builder.start();
    CommandLineThread commandLineThread = new CommandLineThread(process);
    LogStreamReader logStream = new LogStreamReader(process.getInputStream());
    Thread logStreamThread = new Thread(logStream, "LogStreamReader");
    // log collecting thread should be always put first in the executor
    processExecutor.execute(logStreamThread);
    processExecutor.execute(commandLineThread);
    return  new CommandLineThreadWrapper(commandLine, commandLineThread,
      logStreamThread, logStream, process);
  }


  /**
   * Waits for all threads to join that have started python process to tar config files for component
   * @param processExecutor {@link ExecutorService} executes the process when threads are available in the pool
   * @param pythonCmdThreads list of {@link CommandLineThreadWrapper}
   * @param timeout {Integer} time to wait for the threads to join
   * @throws SystemException
   */
  private void waitForAllThreadsToJoin(ExecutorService  processExecutor, List <CommandLineThreadWrapper> pythonCmdThreads, Integer timeout)
    throws SystemException {
    processExecutor.shutdown();
    try {
      processExecutor.awaitTermination(timeout, TimeUnit.MILLISECONDS);
      processExecutor.shutdownNow();
      for (CommandLineThreadWrapper commandLineThreadWrapper: pythonCmdThreads) {
        CommandLineThread commandLineThread = commandLineThreadWrapper.getCommandLineThread();
        try {
          Integer returnCode = commandLineThread.getReturnCode();
          if (returnCode == null) {
            throw new TimeoutException();
          } else if (returnCode != 0) {
            throw new ExecutionException(String.format("Execution of \"%s\" returned %d.", commandLineThreadWrapper.getCommandLine(), returnCode),
              new Throwable(commandLineThreadWrapper.getLogStream().getOutput()));
          }
        } catch (TimeoutException e) {
          LOG.error("Generate client configs script was killed due to timeout ", e);
          throw new SystemException("Generate client configs script was killed due to timeout ", e);
        } catch (ExecutionException e) {
          LOG.error(e.getMessage(), e);
          throw new SystemException(e.getMessage() + " " + e.getCause());
        } finally {
          commandLineThreadWrapper.getProcess().destroy();
        }
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      processExecutor.shutdownNow();
      LOG.error("Failed to run generate client configs script for components");
      throw new SystemException("Failed to run generate client configs script for components");
    }
  }


  /**
   * wrapper class that holds all information and references to the thread and python process
   * started to create compressed configuration config files
   */
  private static class CommandLineThreadWrapper {

    private String commandLine;

    private CommandLineThread commandLineThread;

    private Thread logStreamThread;

    private LogStreamReader logStream;

    private Process process;

    private CommandLineThreadWrapper(String commandLine, CommandLineThread commandLineThread,
                                     Thread logStreamThread, LogStreamReader logStream, Process process) {
      this.commandLine = commandLine;
      this.commandLineThread = commandLineThread;
      this.logStreamThread = logStreamThread;
      this.logStream  =  logStream;
      this.process = process;
    }

    /**
     * Returns commandLine that starts pyton process
     * @return {@link #commandLine}
     */
    private String getCommandLine() {
      return commandLine;
    }

    /**
     * Sets {@link #commandLine}
     * @param commandLine {String}
     */
    private void setCommandLine(String commandLine) {
      this.commandLine = commandLine;
    }

    /**
     * Returns thread that starts and waits for python process to complete
     * @return {@link #commandLineThread}
     */
    private CommandLineThread getCommandLineThread() {
      return commandLineThread;
    }

    /**
     * Sets {@link #commandLineThread}
     * @param commandLineThread {@link CommandLineThread}
     */
    private void setCommandLineThread(CommandLineThread commandLineThread) {
      this.commandLineThread = commandLineThread;
    }

    /**
     * Returns thread that starts and waits to get the output and error log stream from the python process
     * @return {@link #logStreamThread}
     */
    private Thread getLogStreamThread() {
      return logStreamThread;
    }

    /**
     * Sets {@link #logStreamThread}
     * @param logStreamThread {@link Thread}
     */
    private void setLogStreamThread(Thread logStreamThread) {
      this.logStreamThread = logStreamThread;
    }

    /**
     * Returns log stream from the python subprocess
     * @return {@link #logStream}
     */
    private LogStreamReader getLogStream() {
      return logStream;
    }

    /**
     * Sets {@link #logStream}
     * @param logStream {@link LogStreamReader}
     */
    private void setLogStream(LogStreamReader logStream) {
      this.logStream = logStream;
    }

    /**
     * Returns python process
     * @return {@link #process}
     */
    private Process getProcess() {
      return process;
    }

    /**
     *  Sets {@link #process}
     * @param process {@link Process}
     */
    private void setProcess(Process process) {
      this.process = process;
    }
  }

  /**
   * Class to run python process to compress config files as seperate thread
   */
  private static class CommandLineThread extends Thread {
    private final Process process;
    private Integer returnCode;

    private void setReturnCode(Integer exit) {
      returnCode = exit;
    }

    private Integer getReturnCode() {
      return returnCode;
    }

    private CommandLineThread(Process process) {
      this.process = process;
    }


    @Override
    public void run() {
      try {
        setReturnCode(process.waitFor());
      } catch (InterruptedException ignore) {
        return;
      }
    }

  }

  /**
   * Class to collect output and error stream of python subprocess
   */
  private class LogStreamReader implements Runnable {

    private BufferedReader reader;
    private StringBuilder output;

    public LogStreamReader(InputStream is) {
      reader = new BufferedReader(new InputStreamReader(is));
      output = new StringBuilder("");
    }

    public String getOutput() {
      return output.toString();
    }

    @Override
    public void run() {
      try {
        String line = reader.readLine();
        while (line != null) {
          output.append(line);
          output.append("\n");
          line = reader.readLine();
        }
        reader.close();
      } catch (IOException e) {
