使用fastapi和pulumi搭建基于Azure云的IAC Restful API服务

前言

在IAC（即Infrastructure As Code，基础设施即代码）领域，Terraform 是一个老牌工具，使用HCL（HashiCorp Configuration Language）语言来编写配置文件。它支持几乎所有主流的云提供商，如AWS、Azure、GCP等，通过如下几个关键步骤管理基础设施：编写配置文件、初始化环境、生成执行计划、应用计划以及查看状态。

Pulumi 则是一个较新的IAC工具，支持使用多种常见的编程语言（如TypeScript、Python、Go、C#）来编写基础设施代码，而不仅限于声明式语法。这带来了许多灵活性和优势：

使用常用编程语言：Pulumi允许使用者使用熟悉的编程语言编写基础设施配置，这样可以使用语言本身的功能，如条件语句、循环、函数和模块化，提高代码的可读性和重用性。
更好的集成：Pulumi在与现有开发工具链（如CI/CD管道、测试框架等）集成方面表现出色，因此可以使用流行的编程生态系统中的库和工具。这对DevOps团队来说非常有吸引力。
灵活的状态管理：虽然Terraform也有状态文件用于记录资源的现实状态，Pulumi进一步简化了状态管理，允许使用不同的存储后端，包括云存储和Pulumi自己的服务。
更高的可扩展性：Pulumi支持JavaScript、TypeScript、Python、Go和C#等多种语言，这让它能够适应更广泛的需求和团队技术栈，可以更灵活地处理复杂的基础设施场景。

本文使用Python语言，以pulumi作为后端IAC工具，服务端采用FastAPI框架提供Restful能力，来实现基于Azure云的IAC Restful API服务

架构设计

架构说明：

开发人员调用http接口来进行IAC操作
FastAPI服务端分为2个部分，一个gateway负责对外提供restful api服务，一个backend端负责解析前端数据，调用pulumi模块生成IAC代码
任务信息会存入mysql数据库，起到任务追溯的作用以及作为资源cmdb的数据源
IAC操作完成后，会将此次操作变更同步进入资源CMDB

Pulumi模块说明

与经典的命令行模式执行pulumi来进行资源创建/变更不同，在本架构中使用Pulumi Automation来实现IAC能力。Pulumi Automation提供了一种编程接口，使得基础设施管理过程可以完全自动化和自定义。

Pulumi模块支持哪些资源

该模块涉及：

Azure资源模块

Resource GroupAks Cluster(ManagedCluster)Aks Nodepool(AgentPool)Availability SetContainer RegistryData FactoryDatabricks Access ConnectorDatabricks WorkspaceKey VaultLoad BalancerMysql Flexible ServerMysql Single ServerNetwork InterfaceNetwork Security GroupPrivate ZoneRecovery Service VaultRedisRole AssignmentRoute TableStorage AccountSubnetUser Assigned IdentityVirtual Machine

目前我们的pulumi模块现已支持以上azure资源

感兴趣的可以联系我们，提供相关代码模块

主要特点

集成灵活性：利用Pulumi Automation，可以更轻松地将基础设施管理操作嵌入到现有的应用程序或平台中。这对于需要与其他系统进行复杂交互的场景尤为适用，例如动态响应用户请求或事件驱动的基础设施变更。
复用性和模块化：通过将Pulumi的命令行操作封装在代码中，Pulumi Automation允许我们创建更具复用性和模块化的基础设施管理代码库。这对于大型团队和复杂项目尤其有价值，因为代码更易于维护和共享。
状态和并行处理：使用Pulumi Automation，可以更细致地管理资源状态和并行操作。我们可以根据具体情况编写代码来处理不同的资源状态，以及根据依赖关系有序地创建或销毁资源，这可以显著提升执行效率和可靠性。
错误处理和重试机制：通过编写自定义的错误处理和重试逻辑，Pulumi Automation可以更健壮地应对基础设施操作中的各种可能错误。我们可以在代码中捕获异常并进行适当的重试或回滚动作，而不必依赖于外部脚本或手动操作。

依赖模块

需求python 3.7+，执行pip install -r requirements.txt来安装依赖模块

requirements.txt

pulumi>=3.0.0,<4.0.0pulumi-azure-native>=1.0.0,<2.0.0fastapi[all]jsonify>=0.5,<0.6pydantic>=1.9.1,<1.9.2pymysqlpyparsingpulumi_azure

实验步骤

前置条件

Python

需求python 3.7+，执行pip install -r requirements.txt来安装依赖模块

Service Principal

服务主体必须对该订阅具有完全访问权限

pulumi环境变量

必须配置PULUMI_CONFIG_PASSPHRASE供pulumi连接到您的stack，若您使用非本地文件作为backend，则还需配置相应的backend所需字段

mysql数据库（可选）

用以记录资源创建任务信息

资源CMDB（可选）

根据自身需要，可配置一个资源CMDB，存放云资源实时参数配置与状态，以供实际使用

流程说明

FastAPI gateway将异步调用backend方法，传递前端参数给到后端方法
FastAPI后端将参数入库，并调用pulumi模块，生成IAC代码
Pulumi模块内部使用Azure API，进行IAC操作

以下步骤为可选步骤

资源信息同步至cmdb
FastAPI接口提供资源信息查询

步骤一、创建fastapi应用服务端

使用以下代码创建fastapi app（即对外restful api服务）

app.py

@auto_deploy_api.post("/create_resource")
async def create_res(VERABLES_DICT: Item, backgroundTasks: BackgroundTasks):
    VERABLES = VERABLES_DICT.VERABLES
    try:
        # 启动异步任务，因为创建资源耗时较长，http请求不应也无法一直保持
        backgroundTasks.add_task(create_resource, VERABLES, id)
        return {"message": "Task running at backend", "id": id}
    except auto.StackAlreadyExistsError:
        return HTTPException(status_code=409,detail="application already exists")
    except Exception as e:
        return Response(status_code=500, content=repr(e))

步骤二、集成fastapi与pulumi模块

将pulumi模块与fastapi服务集成在同一目录下，文件架构如下：

app.py：fastapi 应用入口文件，定义restful接口

auto_deploy_inline.py：调用pulumi模块，执行IAC操作

pulumi_resources/IAC_resource：存放pulumi模块

create_resource

def create_resource(VERABLES, task_id):
    project_name = VERABLES["GLOBAL_VARS"]["PROJECT_NAME"]
    app_name = VERABLES["GLOBAL_VARS"]["APPLICATION_NAME"]
    env = VERABLES["GLOBAL_VARS"]["ENV"]
    stack_name = "%s-%s-%s-stack" % (project_name, app_name, env)
    try:
        # 此方法处理前端传入的参数，调用pulumi模块，进行资源创建，此处仅以网络资源为例
        def resource_init():
            if "RESOURCE_GROUP" not in VERABLES.keys():
                raise Exception("Resource group is mandatory")
            rg = ResourceGroup(VERABLES["RESOURCE_GROUP"]["NAME"], location)
            if "NETWORK" in VERABLES.keys():
                if "ROUTE_TABLE" in VERABLES["NETWORK"].keys():
                    for rt_ele in VERABLES["NETWORK"]["ROUTE_TABLE"]:
                        RouteTable(rt_ele["name"], location, None, **rt_ele)
                if "NETWORK_SECURITY_GROUP" in VERABLES["NETWORK"].keys():
                    for nsg_ele in VERABLES["NETWORK"]["NETWORK_SECURITY_GROUP"]:
                        NetworkSecurityGroup(nsg_ele["name"], location, None, **nsg_ele)
             
 
        resource_stack = auto.create_or_select_stack(stack_name=stack_name,
                                                     project_name=app_name,
                                                     program=resource_init)
 
 
        update_resource_creation_task_record(data_dict)
        print("Start preview...")
        resource_up_res = resource_stack.preview(on_output=print)
        print("Preview OK...")
        time.sleep(1)
        print("Start up...")
        resource_up_res = resource_stack.up(on_output=print)
        print("Up OK...")
        data_dict["status"] = "successed"
        update_resource_creation_task_record(data_dict)
        return (resource_up_res.stdout)
    except Exception as e:
        data_dict["status"] = "failed"
        data_dict["output"] = repr(e)
        update_resource_creation_task_record(data_dict)

步骤三、启动fastapi应用

cd $YOURPATH/resource_deploy/auto_deploy_fastapi
uvicorn app:auto_deploy_api --host 0.0.0.0 --port 8000

步骤四、Postman调用接口创建资源

接下来，我们就可以通过http请求的方式，调用pulumi的IAC能力，操作Azure云的资源了

示例payload

payload

{
    'VERABLES': {
        'GLOBAL_VARS': {
            'PROJECT_NAME': 'auto',
            'APPLICATION_NAME': 'auto',
            'ENV': 'dev',
            'LOCATION': 'chinanorth3',
            'SUBSCRIPTION_ID': '***********'
        },
        'RESOURCE_GROUP': {
            'NAME': 'test-pulumi-rg',
            'TAGS': {
                'ApplicationID': 'ITPlatform',
                'ApplicationName': 'ITPlatform'
            }
        },
        'NETWORK': {
            'ROUTE_TABLE': [{
                'name': 'test-rt01',
                'resource_group_name': 'test-network-rg01',
                'routes': [{
                    'name': 'rule1',
                    'address_prefix': '0.0.0.0/0',
                    'next_hop_type': 'VirtualAppliance',
                    'next_hop_ip_address': '10.20.128.140'
                }]
            }],
            'NETWORK_SECURITY_GROUP': [{
                'name': 'test-nsg01',
                'resource_group_name': 'test-network-rg01',
                'rules': [{
                    'access': 'Allow',
                    'destination_address_prefix': '*',
                    'destination_port_range': '*',
                    'direction': 'Inbound',
                    'name': 'rule1',
                    'priority': 1001,
                    'protocol': '*',
                    'source_address_prefixes': ['10.20.2.0/24'],
                    'source_port_range': '*'
                }]
            }]
        }
}

附言

用类似的方法，可以创建删除资源的接口与查看任务状态的接口，示例如下：

app.py

@auto_deploy_api.delete("/delete_resource")
async def delete_res(VERABLES_DICT: Item, backgroundTasks: BackgroundTasks):
    VERABLES = VERABLES_DICT.VERABLES
    try:
        # 启动异步任务，因为删除资源耗时较长，http请求不应也无法一直保持
        backgroundTasks.add_task(delete_resource, VERABLES)
        return {"message": "Will delete"}
    except auto.StackNotFoundError:
        return HTTPException(status_code=404,detail="application not found")
    except Exception as e:
        return Response(status_code=500, content=repr(e))
 
@auto_deploy_api.post("/stack_status")
async def delete_res(VERABLES_DICT: Item):
    VERABLES = VERABLES_DICT.VERABLES
    try:
        data = get_stack_status(VERABLES["ID"])
        return data
    except auto.StackNotFoundError:
        return HTTPException(status_code=404,detail="id not found")
    except Exception as e:
        return Response(status_code=500, content=repr(e))

auto_deploy_inline.py

def delete_resource(VERABLES):
    stack_name = VERABLES["GLOBAL_VARS"]["STACK_NAME"]
    project_name = VERABLES["GLOBAL_VARS"]["PROJECT_NAME"]
    client_id, client_secret, client_tenant = get_azure_sp()
    def resource_init():
        pass
 
    resource_stack = auto.create_or_select_stack(stack_name=stack_name,
                                                 project_name=app_name,
                                                 program=resource_init)
    resource_stack.refresh(on_output=print)
    resource_destroy_res = resource_stack.destroy(on_output=print)
  
    return (resource_destroy_res.stdout)
 
 
def get_stack_status(id):
    data = get_task_record(id)
    if len(data) == 0:
        raise auto.StackNotFoundError
    data_dict = {"id": id, "stack_name": data[0][1], "status": data[0][3],
                 "payload": data[0][2], "output": data[0][4]}
 
    return data_dict

调用方式：