Use ProcessPool rather than ThreadPool Executor
[ccsdk/cds.git] / ms / command-executor / src / main / python / command_executor_handler.py
1 #
2 # Copyright (C) 2019 Bell Canada.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #      http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 from builtins import Exception, open, dict
17 from subprocess import CalledProcessError, PIPE
18 from google.protobuf.json_format import MessageToJson
19
20 import logging
21 import os
22 import re
23 import subprocess
24 import virtualenv
25 import venv
26 import utils
27 import proto.CommandExecutor_pb2 as CommandExecutor_pb2
28
29 REQUIREMENTS_TXT = "requirements.txt"
30
31
32 class CommandExecutorHandler():
33
34     def __init__(self, request):
35         self.request = request
36         self.logger = logging.getLogger(self.__class__.__name__)
37         self.blueprint_id = utils.get_blueprint_id(request)
38         # FIXME parameterize path
39         self.venv_home = '/opt/app/onap/blueprints/deploy/' + self.blueprint_id
40         self.installed = self.venv_home + '/.installed'
41
42     def is_installed(self):
43         return os.path.exists(self.installed)
44
45     def prepare_env(self, request, results):
46         if not self.is_installed():
47             self.create_venv()
48             if not self.activate_venv():
49                 return False
50
51             f = open(self.installed, "w+")
52             if not self.install_packages(request, CommandExecutor_pb2.pip, f, results):
53                 return False
54             f.write("\r\n")
55             results.append("\n")
56             if not self.install_packages(request, CommandExecutor_pb2.ansible_galaxy, f, results):
57                 return False
58             f.close()
59         else:
60             f = open(self.installed, "r")
61             results.append(f.read())
62             f.close()
63
64         return True
65
66     def execute_command(self, request, results):
67
68         if not self.activate_venv():
69             return False
70
71         cmd = "cd " + self.venv_home
72
73         if "ansible-playbook" in request.command:
74             cmd = cmd + "; " + request.command + " -e 'ansible_python_interpreter=" + self.venv_home + "/bin/python'"
75         else:
76             # we append the properties as last agr to the script
77             cmd = cmd + "; " + request.command + " " + re.escape(MessageToJson(request.properties))
78
79         try:
80             with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
81                                   shell=True, bufsize=1, universal_newlines=True) as newProcess:
82                 while True:
83                     output = newProcess.stdout.readline()
84                     if output == '' and newProcess.poll() is not None:
85                         break
86                     if output:
87                         self.logger.info(output.strip())
88                         results.append(output.strip())
89                     rc = newProcess.poll()
90         except Exception as e:
91             self.logger.info("{} - Failed to execute command. Error: {}".format(self.blueprint_id, e))
92             results.append(e)
93             return False
94
95         # deactivate_venv(blueprint_id)
96         return True
97
98     def install_packages(self, request, type, f, results):
99         for package in request.packages:
100             if package.type == type:
101                 f.write("Installed %s packages:\r\n" % CommandExecutor_pb2.PackageType.Name(type))
102                 for p in package.package:
103                     f.write("   %s\r\n" % p)
104                     if package.type == CommandExecutor_pb2.pip:
105                         success = self.install_python_packages(p, results)
106                     else:
107                         success = self.install_ansible_packages(p, results)
108                     if not success:
109                         f.close()
110                         os.remove(self.installed)
111                         return False
112         return True
113
114     def install_python_packages(self, package, results):
115         self.logger.info(
116             "{} - Install Python package({}) in Python Virtual Environment".format(self.blueprint_id, package))
117
118         if REQUIREMENTS_TXT == package:
119             command = ["pip", "install", "-r", self.venv_home + "/Environments/" + REQUIREMENTS_TXT]
120         else:
121             command = ["pip", "install", package]
122
123         env = dict(os.environ)
124         if "https_proxy" in os.environ:
125             env['https_proxy'] = os.environ['https_proxy']
126
127         try:
128             results.append(subprocess.run(command, check=True, stdout=PIPE, stderr=PIPE, env=env).stdout.decode())
129             results.append("\n")
130             return True
131         except CalledProcessError as e:
132             results.append(e.stderr.decode())
133             return False
134
135     def install_ansible_packages(self, package, results):
136         self.logger.info(
137             "{} - Install Ansible Role package({}) in Python Virtual Environment".format(self.blueprint_id, package))
138         command = ["ansible-galaxy", "install", package, "-p", self.venv_home + "/Scripts/ansible/roles"]
139
140         env = dict(os.environ)
141         if "http_proxy" in os.environ:
142             # ansible galaxy uses https_proxy environment variable, but requires it to be set with http proxy value.
143             env['https_proxy'] = os.environ['http_proxy']
144
145         try:
146             results.append(subprocess.run(command, check=True, stdout=PIPE, stderr=PIPE, env=env).stdout.decode())
147             results.append("\n")
148             return True
149         except CalledProcessError as e:
150             results.append(e.stderr.decode())
151             return False
152
153     def create_venv(self):
154         self.logger.info("{} - Create Python Virtual Environment".format(self.blueprint_id))
155         try:
156             bin_dir = self.venv_home + "/bin"
157             # venv doesn't populate the activate_this.py script, hence we use from virtualenv
158             venv.create(self.venv_home, with_pip=True, system_site_packages=True)
159             virtualenv.writefile(os.path.join(bin_dir, "activate_this.py"), virtualenv.ACTIVATE_THIS)
160         except Exception as err:
161             self.logger.info(
162                 "{} - Failed to provision Python Virtual Environment. Error: {}".format(self.blueprint_id, err))
163
164     def activate_venv(self):
165         self.logger.info("{} - Activate Python Virtual Environment".format(self.blueprint_id))
166
167         # Fix: The python generated activate_this.py script concatenates the env bin dir to PATH on every call
168         #      eventually this process PATH variable was so big (128Kb) that no child process could be spawn
169         #      This script will remove all duplicates; while keeping the order of the PATH folders
170         fixpathenvvar = "os.environ['PATH']=os.pathsep.join(list(dict.fromkeys(os.environ['PATH'].split(':'))))"
171
172         path = "%s/bin/activate_this.py" % self.venv_home
173         try:
174             exec(open(path).read(), {'__file__': path})
175             exec(fixpathenvvar)
176             self.logger.info("Running with PATH : {}".format(os.environ['PATH']))
177             return True
178         except Exception as err:
179             self.logger.info(
180                 "{} - Failed to activate Python Virtual Environment. Error: {}".format(self.blueprint_id, err))
181             return False
182
183     def deactivate_venv(self):
184         self.logger.info("{} - Deactivate Python Virtual Environment".format(self.blueprint_id))
185         command = ["deactivate"]
186         try:
187             subprocess.run(command, check=True)
188         except Exception as err:
189             self.logger.info(
190                 "{} - Failed to deactivate Python Virtual Environment. Error: {}".format(self.blueprint_id, err))