2.6. 在 Microsoft Azure 中创建功能来过滤您的数据
创建过滤数据的功能,并将其添加到您创建的要与红帽共享的存储帐户中。您可以使用示例 Python 脚本从与红帽费用相关的成本导出中收集成本数据,并将其添加到存储帐户中。
先决条件
- 在您的设备上必须安装 Visual Studio Code。
- 您必须在 Visual Studio Code 中安装 Microsoft Azure 功能扩展。
流程
- 登录到您的 Microsoft Azure 帐户。
-
在搜索栏中输入
functions
,选择 Functions,然后单击 Create。 - 为您的功能选择一个托管选项,然后点 。
在 Create Function App 页面中,通过添加资源组来配置您的功能应用程序。
- 在 Instance Details 部分中,将功能命名为 app。
- 在 Runtime stack 中,选择 Python
- 在 Version 中,选择 3.10。
点 Review + create:
- 点 Create。
- 点 Go to resource 来配置函数。
在功能应用程序菜单中,点击 Functions 来创建时间触发器功能:
- 点 Create。
- 在开发环境字段中,选择 VSCode。
打开 Visual Studio Code,并确保安装了 Microsoft Azure Functions Visual Studio Code 扩展。要创建 Azure 功能,Microsoft 建议使用 Microsoft Visual Studio Code IDE 来开发和部署代码。有关配置 Visual Studio Code 的更多信息,请参阅 Quickstart: 使用 Visual Studio Code 在 Azure 中使用 Python 创建功能 。
- 点 Visual Studio Code 中的 Microsoft Azure 选项卡,登录到 Azure。
- 在 Visual Studio Code 的 Workspaces 选项卡中,点 Create function。
- 按照提示为您的功能设置本地位置,并为功能选择语言和版本。在本例中,选择 Python,对于 并选择 Python 3.9。
- 在 Select a template for your project first function 对话框中,选择 Timer trigger,命名功能,然后按 Enter
-
为希望该函数运行时设置 cron 表达式。在本例中,使用
0*9***
在每天上午 9 点运行该功能。 - 点 Create。
在开发环境中创建功能后,打开
requirements.txt
文件,添加以下要求并保存文件:azure-functions pandas requests azure-identity azure-storage-blob
打开
__init__.py
并粘贴以下 Python 脚本。将标记为# Required vars 的部分中
的值更改为您的环境的值。对于USER
和PASS
值,您可以选择使用 Key Vault 凭据 将用户名和密码配置为环境变量。import datetime import logging import uuid import requests import pandas as pd from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient, ContainerClient import azure.functions as func def main(mytimer: func.TimerRequest) -> None: utc_timestamp = datetime.datetime.utcnow().replace( tzinfo=datetime.timezone.utc).isoformat() default_credential = DefaultAzureCredential() now = datetime.datetime.now() year = now.strftime("%Y") month = now.strftime("%m") day = now.strftime("%d") output_blob_name=f"{year}/{month}/{day}/{uuid.uuid4()}.csv" # Required vars to update USER = os.getenv('UsernameFromVault') # Cost management username PASS = os.getenv('PasswordFromVault') # Cost management password integration_id = "<your_integration_id>" # Cost management integration_id cost_export_store = "https://<your-cost-export-storage-account>.blob.core.windows.net" # Cost export storage account url cost_export_container = "<your-cost-export-container>" # Cost export container filtered_data_store = "https://<your_filtered_data_container-storage-account>.blob.core.windows.net" # Filtered data storage account url filtered_data_container = "<your_filtered_data_container>" # Filtered data container # Create the BlobServiceClient object blob_service_client = BlobServiceClient(filtered_data_store, credential=default_credential) container_client = ContainerClient(cost_export_store, credential=default_credential, container_name=cost_export_container) blob_list = container_client.list_blobs() latest_blob = None for blob in blob_list: if latest_blob: if blob.last_modified > latest_blob.last_modified: latest_blob = blob else: latest_blob = blob bc = container_client.get_blob_client(blob=latest_blob) data = bc.download_blob() blobjct = "/tmp/blob.csv" with open(blobjct, "wb") as f: data.readinto(f) df = pd.read_csv(blobjct) filtered_data = df.loc[((df["publisherType"] == "Marketplace") & ((df["publisherName"].astype(str).str.contains("Red Hat")) | (((df["publisherName"] == "Microsoft") | (df["publisherName"] == "Azure")) & (df['meterSubCategory'].astype(str).str.contains("Red Hat") | df['serviceInfo2'].astype(str).str.contains("Red Hat")))))] filtered_data_csv = filtered_data.to_csv (index_label="idx", encoding = "utf-8") blob_client = blob_service_client.get_blob_client(container=filtered_data_container, blob=output_blob_name) blob_client.upload_blob(filtered_data_csv, overwrite=True) # Post results to console.redhat.com API url = "https://console.redhat.com/api/cost-management/v1/ingress/reports/" json_data = {"source": integration_id, "reports_list": [f"{filtered_data_container}/{output_blob_name}"], "bill_year": year, "bill_month": month} resp = requests.post(url, json=json_data, auth=(USER, PASS)) logging.info(f'Post result: {resp}') if mytimer.past_due: logging.info('The timer is past due!') logging.info('Python timer trigger function ran at %s', utc_timestamp)
- 保存该文件。
- 将功能部署到 Microsoft Azure。