def launch_command()

in azurelinuxagent/ga/exthandlers.py [0:0]
56 lines of code
9 McCabe index (conditional complexity)

    def launch_command(self, cmd, cmd_name=None, timeout=300, extension_error_code=ExtensionErrorCodes.PluginProcessingError,
                       env=None, extension=None):
        begin_utc = datetime.datetime.utcnow()
        self.logger.verbose("Launch command: [{0}]", cmd)

        base_dir = self.get_base_dir()

        with tempfile.TemporaryFile(dir=base_dir, mode="w+b") as stdout:
            with tempfile.TemporaryFile(dir=base_dir, mode="w+b") as stderr:
                if env is None:
                    env = {}

                # Always add Extension Path and version to the current launch_command (Ask from publishers)
                env.update({
                    ExtCommandEnvVariable.ExtensionPath: base_dir,
                    ExtCommandEnvVariable.ExtensionVersion: str(self.ext_handler.version),
                    ExtCommandEnvVariable.WireProtocolAddress: self.protocol.get_endpoint(),

                    # Setting sequence number to 0 incase no settings provided to keep in accordance with the empty
                    # 0.settings file that we create for such extensions.
                    ExtCommandEnvVariable.ExtensionSeqNumber: str(
                        extension.sequenceNumber) if extension is not None else _DEFAULT_SEQ_NO
                })

                if self.should_perform_multi_config_op(extension):
                    env[ExtCommandEnvVariable.ExtensionName] = extension.name

                supported_features = []
                for _, feature in get_agent_supported_features_list_for_extensions().items():
                    supported_features.append(
                        {
                            "Key": feature.name,
                            "Value": feature.version
                        }
                    )
                if supported_features:
                    env[ExtCommandEnvVariable.ExtensionSupportedFeatures] = json.dumps(supported_features)

                ext_name = self.get_extension_full_name(extension)
                try:
                    # Some extensions erroneously begin cmd with a slash; don't interpret those
                    # as root-relative. (Issue #1170)
                    command_full_path = os.path.join(base_dir, cmd.lstrip(os.path.sep))
                    log_msg = "Executing command: {0} with environment variables: {1}".format(command_full_path,
                                                                                              json.dumps(env))
                    self.logger.info(log_msg)
                    self.report_event(name=ext_name, message=log_msg, log_event=False)

                    # Add the os environment variables before executing command
                    env.update(os.environ)
                    process_output = CGroupConfigurator.get_instance().start_extension_command(
                        extension_name=self.get_full_name(extension),
                        command=command_full_path,
                        cmd_name=cmd_name,
                        timeout=timeout,
                        shell=True,
                        cwd=base_dir,
                        env=env,
                        stdout=stdout,
                        stderr=stderr,
                        error_code=extension_error_code)

                except OSError as e:
                    raise ExtensionError("Failed to launch '{0}': {1}".format(command_full_path, e.strerror),
                                         code=extension_error_code)

                duration = elapsed_milliseconds(begin_utc)
                log_msg = "Command: {0}\n{1}".format(cmd, "\n".join(
                    [line for line in process_output.split('\n') if line != ""]))
                self.logger.info(log_msg)
                self.report_event(name=ext_name, message=log_msg, duration=duration, log_event=False)

                return process_output