如何在 Python 中运行 Google Cloud Function 中的子进程 [英] How to run a subprocess inside Google Cloud Function in Python

问题描述

我试图在 GCP 函数中运行一个 bash 脚本，但不知何故它不起作用.这是我的函数，它基本上将文件(代理)导出到 Google Apigee:

def test2(请求):cmd = "python ./my-proxy/tools/deploy.py -n myProxy -u userName:!password -o myOrg -e test -d ./my-proxy -p/"# 没有阻塞，它启动一个子进程.p = subprocess.Popen(cmd，shell=True，stdout=subprocess.PIPE，stderr=subprocess.PIPE)# 你可以阻止 util cmd 执行完成p.wait()# 或标准输出，标准错误 = p.communicate()返回代理部署到 Apigee"

这是我的 deploy.py 文件的样子:

!/usr/bin/env python导入base64导入getopt导入 httplib导入json进口重新导入操作系统导入系统导入字符串IO导入网址解析导入 xml.dom.minidom导入压缩文件def httpCall(动词，uri，标题，正文):如果 httpScheme == 'https':conn = httplib.HTTPSConnection(httpHost)别的:conn = httplib.HTTPConnection(httpHost)如果标题 == 无:hdrs = 字典()别的:hdrs = 标题hdrs['Authorization'] = '基本 %s' % base64.b64encode(UserPW)conn.request(动词，uri，正文，hdrs)返回 conn.getresponse()定义 getElementText(n):c = n.firstChildstr = StringIO.StringIO()而 c != 无:如果 c.nodeType == xml.dom.Node.TEXT_NODE:str.write(c.data)c = c.nextSibling返回 str.getvalue().strip()def getElementVal(n，名称):c = n.firstChild而 c != 无:如果 c.nodeName == 名称:返回 getElementText(c)c = c.nextSibling返回无# 如果文件路径的任何组成部分包含一个目录名，则返回 TRUE# 以."开头像'.svn'，但不是'.'或者 '..'def pathContainsDot(p):c = re.compile('.w+')对于 p.split('/') 中的 pc:如果 c.match(pc) != 无:返回真返回假定义 getDeployments():# 打印部署信息hdrs = {'接受':'应用程序/xml'}resp = httpCall('GET','/v1/organizations/%s/apis/%s/deployments' ％ (机构名称)，hdrs，无)如果 resp.status != 200:返回无ret = 列表()部署 = xml.dom.minidom.parse(resp)环境 = deployments.getElementsByTagName('Environment')对于环境中的 env:envName = env.getAttribute('name')修订版 = env.getElementsByTagName('修订版')对于修订版本:revNum = int(rev.getAttribute('name'))错误 = 无state = getElementVal(rev, 'State')basePaths = rev.getElementsByTagName('BasePath')如果 len(basePaths) >0:basePath = getElementText(basePaths[0])别的:basePath = '未知'# svrs = rev.getElementsByTagName('Server')状态 = {'环境':环境名称，修订":revNum，'basePath':基本路径，状态":状态}如果错误！= 无:状态['错误'] = 错误ret.append(状态)返回 retdef printDeployments(dep):d 在 dep 中:print '环境:%s' % d['environment']print ' 修订: %i BasePath = %s' % (d['revision'], d['basePath'])print '状态:%s' % d['state']如果 d 中出现错误":print '错误: %s' % d['error']ApigeeHost = 'https://api.enterprise.apigee.com'用户密码 = 无目录 = 无组织 = 无环境 = 无名称 = 无BasePath = '/'应该部署 = 真选项 = 'h:u:d:e:n:p:o:i:z:'opts = getopt.getopt(sys.argv[1:], 选项)[0]对于 o 在 opts 中:如果 o[0] == '-n':名称 = o[1]elif o[0] == '-o':组织 = o[1]elif o[0] == '-h':ApigeeHost = o[1]elif o[0] == '-d':目录 = o[1]elif o[0] == '-e':环境 = o[1]elif o[0] == '-p':BasePath = o[1]elif o[0] == '-u':用户密码 = o[1]elif o[0] == '-i':应该部署 = 假elif o[0] == '-z':压缩文件 = o[1]如果 UserPW == None 或 (目录 == 无和 ZipFile == 无)或 环境 == 无或 名称 == 无或 组织 == 无:print """用法:deploy -n [name] (-d [目录名] | -z [zipfile])-e [环境] -u [用户名:密码] -o [组织][-p [基本路径] -h [apigee API 网址] -i]基本路径默认为/"Apigee 网址默认为https://api.enterprise.apigee.com"-i 表示仅导入而不实际部署"""系统退出(1)url = urlparse.urlparse(ApigeeHost)httpScheme = 网址[0]httpHost = 网址[1]身体=无如果目录！=无:# 在内存中构造一个压缩包的副本tf = StringIO.StringIO()zipout = zipfile.ZipFile(tf, 'w')dirList = os.walk(目录)对于 dirList 中的 dirEntry:如果不是 pathContainsDot(dirEntry[0]):对于 dirEntry[2] 中的文件条目:如果不是 fileEntry.endswith('~'):fn = os.path.join(dirEntry[0], fileEntry)en = os.path.join(os.path.relpath(dirEntry[0], 目录),文件入口)print '正在将 %s 写入 %s' % (fn, en)zipout.write(fn, en)zipout.close()身体 = tf.getvalue()elif ZipFile != 无:f = 打开(ZipFile，'r')正文 = f.read()f.close()# 将包上传到 APIhdrs = {'Content-Type': '应用程序/八位字节流','接受':'应用程序/json'}uri = '/v1/organizations/%s/apis?action=import&name=%s' % (机构名称)resp = httpCall('POST', uri, hdrs, body)如果 resp.status != 200 和 resp.status != 201:print '导入到 %s 失败，状态为 %i:
%s' % (uri, resp.status, resp.read())系统退出(2)部署 = json.load(resp)修订= int(部署['修订'])print '导入的新代理版本 %i' % 修订如果应该部署:# 取消部署副本deps = getDeployments()对于 d 在部门:如果 d['environment'] == 环境和 d['basePath'] == BasePath 和 d['revision'] != 修订:print '在相同的环境和路径中取消部署修订 %i:' % d['修订']conn = httplib.HTTPSConnection(httpHost)resp = httpCall('POST',('/v1/organizations/%s/apis/%s/deployments' +'?action=取消部署' +'&env=%s' +'&revision=%i') % (组织，名称，环境，d['revision'])，无，无)如果 resp.status != 200 和 resp.status != 204:print '错误 %i 取消部署:
%s' % (resp.status, resp.read())# 部署包hdrs = {'接受':'应用程序/json'}resp = httpCall('POST',('/v1/organizations/%s/apis/%s/deployments' +'?action=deploy' +'&env=%s' +'&修订=%i' +'&basepath=%s') % (组织、名称、环境、修订、BasePath)，hdrs，无)如果 resp.status != 200 和 resp.status != 201:print '部署失败，状态为 %i:
%s' % (resp.status, resp.read())系统退出(2)deps = getDeployments()打印部署(部门)

当我在我的机器上本地运行时，这有效，但在 GCP 上无效.不知道是否与我使用此功能连接到 Google Apigee 的事实有关.奇怪的是 GCP 上的日志没有显示任何错误，但是我没有将代理导出到 Apigee.

感谢帮助！

更新:尝试使用 subprocess.check_output() 受到一些人的鼓励:

def 测试(请求):输出 = 无尝试:输出 = subprocess.check_output(["./my-proxy/tools/deploy.py",'-n', 'myProxy','-u'，'我的用户名:我的密码'，'-o', '我的组织名称','-e'，'测试'，'-d', './my-proxy','-p', '/'])除了:打印(输出)返回输出

仍然无法在 GCP 上工作.就像我之前提到的，它在我的机器中就像一个魅力(以上两种解决方案)，但在 GCP 中却没有.如下图所示，从 GCP 执行 deploy.py 后我得到了 200，但我的文件没有转到 Apigee:

GCP 日志也没有显示任何错误:

解决方案

这是可能的！

python 可执行文件未在 Cloud Function 运行时中安装或链接，但 python3 已安装或链接.因此，有几种方法可以解决这个问题:

1. 将 python3 指定为要运行的程序:"python3 ./my-proxy/tools/deploy.py ...";p>

2. 在 deploy.py 脚本中添加 #! 运算符:#!/usr/bin/env python3;

3. 将python解释器指定为Popen.您可以使用 sys.executable 来引用当前使用的可执行文件:

    进程 = subprocess.Popen([sys.executable,./deploy.py",-n"，我的代理"，-u"，我的用户名:我的密码"，-o"，我的组织名称"，-e"，测试"，-d"，./my-proxy",-p"，/",],标准输出=子进程.PIPE，stderr=subprocess.PIPE,Universal_newlines=真，)

您没有看到错误，因为它是在子进程中生成的，打印到其 stderr，然后由您的程序使用 process.communicate() 或 process.check_output(...)，但未打印.要查看您遇到的错误，您可以打印出 stdout 和 stderr 的内容:

 out, err = process.communicate()log.debug("returncode = %s", process.returncode)log.debug("stdout = %s", out)log.debug("stderr = %s", err)

在 github

I'm trying to run a bash script inside GCP Function but somehow it's not working. Here my function, which basically export a file(proxy) to Google Apigee:

def test2(request):
    cmd = "python ./my-proxy/tools/deploy.py -n myProxy -u userName:!password -o myOrg -e test -d ./my-proxy -p /"
    # no block, it start a sub process.
    p = subprocess.Popen(cmd , shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    # and you can block util the cmd execute finish
    p.wait()
    # or stdout, stderr = p.communicate()
    return "Proxy deployed to Apigee"

Here's my deploy.py file looks like:

!/usr/bin/env python

import base64
import getopt
import httplib
import json
import re
import os
import sys
import StringIO
import urlparse
import xml.dom.minidom
import zipfile


def httpCall(verb, uri, headers, body):
    if httpScheme == 'https':
        conn = httplib.HTTPSConnection(httpHost)
    else:
        conn = httplib.HTTPConnection(httpHost)

    if headers == None:
        hdrs = dict()
    else:
        hdrs = headers

    hdrs['Authorization'] = 'Basic %s' % base64.b64encode(UserPW)
    conn.request(verb, uri, body, hdrs)

    return conn.getresponse()


def getElementText(n):
    c = n.firstChild
    str = StringIO.StringIO()

    while c != None:
        if c.nodeType == xml.dom.Node.TEXT_NODE:
            str.write(c.data)
        c = c.nextSibling

    return str.getvalue().strip()


def getElementVal(n, name):
    c = n.firstChild

    while c != None:
        if c.nodeName == name:
            return getElementText(c)
        c = c.nextSibling

    return None


# Return TRUE if any component of the file path contains a directory name that
# starts with a "." like '.svn', but not '.' or '..'
def pathContainsDot(p):
    c = re.compile('.w+')

    for pc in p.split('/'):
        if c.match(pc) != None:
            return True

    return False


def getDeployments():
    # Print info on deployments
    hdrs = {'Accept': 'application/xml'}
    resp = httpCall('GET',
            '/v1/organizations/%s/apis/%s/deployments' 
                % (Organization, Name),
            hdrs, None)

    if resp.status != 200:
        return None

    ret = list()
    deployments = xml.dom.minidom.parse(resp)
    environments = deployments.getElementsByTagName('Environment')

    for env in environments:
        envName = env.getAttribute('name')
        revisions = env.getElementsByTagName('Revision')
        for rev in revisions:
            revNum = int(rev.getAttribute('name'))
            error = None
            state = getElementVal(rev, 'State')
            basePaths = rev.getElementsByTagName('BasePath')

            if len(basePaths) > 0:
                basePath = getElementText(basePaths[0])
            else:
                basePath = 'unknown'

            # svrs = rev.getElementsByTagName('Server')
            status = {'environment': envName,
                    'revision': revNum,
                    'basePath': basePath,
                    'state': state}

            if error != None:
                status['error'] = error

            ret.append(status)

    return ret


def printDeployments(dep):
    for d in dep:
        print 'Environment: %s' % d['environment']
        print '  Revision: %i BasePath = %s' % (d['revision'], d['basePath'])
        print '  State: %s' % d['state']
        if 'error' in d:
            print '  Error: %s' % d['error']

ApigeeHost = 'https://api.enterprise.apigee.com'
UserPW = None
Directory = None
Organization = None
Environment = None
Name = None
BasePath = '/'
ShouldDeploy = True

Options = 'h:u:d:e:n:p:o:i:z:'

opts = getopt.getopt(sys.argv[1:], Options)[0]

for o in opts:
    if o[0] == '-n':
        Name = o[1]
    elif o[0] == '-o':
        Organization = o[1]
    elif o[0] == '-h':
        ApigeeHost = o[1]
    elif o[0] == '-d':
        Directory = o[1]
    elif o[0] == '-e':
        Environment = o[1]
    elif o[0] == '-p':
        BasePath = o[1]
    elif o[0] == '-u':
        UserPW = o[1]
    elif o[0] == '-i':
        ShouldDeploy = False
    elif o[0] == '-z':
        ZipFile = o[1]

if UserPW == None or 
        (Directory == None and ZipFile == None) or 
        Environment == None or 
        Name == None or 
        Organization == None:
    print """Usage: deploy -n [name] (-d [directory name] | -z [zipfile])
              -e [environment] -u [username:password] -o [organization]
              [-p [base path] -h [apigee API url] -i]
    base path defaults to "/"
    Apigee URL defaults to "https://api.enterprise.apigee.com"
    -i denotes to import only and not actually deploy
    """
    sys.exit(1)

url = urlparse.urlparse(ApigeeHost)
httpScheme = url[0]
httpHost = url[1]

body = None

if Directory != None:
    # Construct a ZIPped copy of the bundle in memory
    tf = StringIO.StringIO()
    zipout = zipfile.ZipFile(tf, 'w')

    dirList = os.walk(Directory)
    for dirEntry in dirList:
        if not pathContainsDot(dirEntry[0]):
            for fileEntry in dirEntry[2]:
                if not fileEntry.endswith('~'):
                    fn = os.path.join(dirEntry[0], fileEntry)
                    en = os.path.join(
                            os.path.relpath(dirEntry[0], Directory),
                            fileEntry)
                    print 'Writing %s to %s' % (fn, en)
                    zipout.write(fn, en)

    zipout.close()
    body = tf.getvalue()
elif ZipFile != None:
    f = open(ZipFile, 'r')
    body = f.read()
    f.close()

# Upload the bundle to the API
hdrs = {'Content-Type': 'application/octet-stream',
        'Accept': 'application/json'}
uri = '/v1/organizations/%s/apis?action=import&name=%s' % 
            (Organization, Name)
resp = httpCall('POST', uri, hdrs, body)

if resp.status != 200 and resp.status != 201:
    print 'Import failed to %s with status %i:
%s' % 
            (uri, resp.status, resp.read())
    sys.exit(2)

deployment = json.load(resp)
revision = int(deployment['revision'])

print 'Imported new proxy version %i' % revision

if ShouldDeploy:
    # Undeploy duplicates
    deps = getDeployments()
    for d in deps:
        if d['environment'] == Environment and 
            d['basePath'] == BasePath and 
            d['revision'] != revision:
            print 'Undeploying revision %i in same environment and path:' % 
                    d['revision']
            conn = httplib.HTTPSConnection(httpHost)
            resp = httpCall('POST',
                    ('/v1/organizations/%s/apis/%s/deployments' +
                            '?action=undeploy' +
                            '&env=%s' +
                            '&revision=%i') % 
                        (Organization, Name, Environment, d['revision']),
                 None, None)
            if resp.status != 200 and resp.status != 204:
                print 'Error %i on undeployment:
%s' % 
                        (resp.status, resp.read())

    # Deploy the bundle
    hdrs = {'Accept': 'application/json'}
    resp = httpCall('POST',
        ('/v1/organizations/%s/apis/%s/deployments' +
                '?action=deploy' +
                '&env=%s' +
                '&revision=%i' +
                '&basepath=%s') % 
            (Organization, Name, Environment, revision, BasePath),
        hdrs, None)

    if resp.status != 200 and resp.status != 201:
        print 'Deploy failed with status %i:
%s' % (resp.status, resp.read())
        sys.exit(2)

deps = getDeployments()
printDeployments(deps)

This works when I run locally on my machine, but not on GCP. Don't know if have anything to do with the fact I'm connecting to Google Apigee with this function. It's weird that the logs on GCP doesn't show any error, however I don't have my proxy exported to Apigee.

Thanks the help!

UPDATED: tried using subprocess.check_output() as encouraged by some here:

def test(request):
    output = None
    try:
        output = subprocess.check_output([
        "./my-proxy/tools/deploy.py", 
        '-n',  'myProxy',
        '-u', 'myUserName:myPassword',
        '-o', 'myOrgName',
        '-e', 'test',
        '-d', './my-proxy',
        '-p', '/'])

    except:
        print(output)    

    return output 

And still not working on GCP. Like I mentioned before, it works like a charm (both solutions above) in my machine, but in GCP doesn't. As you can see from the image below, I get a 200 after executing deploy.py from GCP but my file doesn't go to Apigee:

GCP logs doesn't show any error as well:

解决方案

This is possible!

The python executable is not installed or linked in the Cloud Function runtime, but python3 is. Hence, there are a few ways to solve this:

1. specify python3 as the program to run: "python3 ./my-proxy/tools/deploy.py ...";

2. add the #! operator in the deploy.py script: #!/usr/bin/env python3;

3. specify the python interpreter to Popen. You can use sys.executable to refer to the currently used executable:

    process = subprocess.Popen(
        [
            sys.executable,
            "./deploy.py",
            "-n",
            "myProxy",
            "-u",
            "myUserName:myPassword",
            "-o",
            "myOrgName",
            "-e",
            "test",
            "-d",
            "./my-proxy",
            "-p",
            "/",
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
    )
   

You did not see an error because it was generated in a subprocess, printed to its stderr, and subsequently captured by your program with process.communicate() or process.check_output(...), but not printed. To see the error you are experiencing, you can print out the content of stdout and stderr:

    out, err = process.communicate()
    log.debug("returncode = %s", process.returncode)
    log.debug("stdout = %s", out)
    log.debug("stderr = %s", err)

Check out our source code we used to analyze, reproduce and solve your question on github

这篇关于如何在 Python 中运行 Google Cloud Function 中的子进程的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！