I'm currently attempting to run a Python script in Azure Runbook on .py runtime 3.8
. The purpose is to monitor the status of a VM and pings an email (not included in the MRE) along with generating a GUID through the use of a decorator calling the generate_incident_reference_number()
method which will return a UUID4
.
Having attained tenant_id
, the cliend_id
, and well as the cliend_secret
via the Home > App Registrations
& registed a Service Principal, I have also checked the Azure SDK
documentation resources & uploaded the following .whl
files to my Automation Account > Shared Resources > Python Packages
:
azure_core 1.32.0
,
azure_identity 1.19.0
,
azure_mgmt_compute 34.0.0
,
azure_mgmt_core 1.5.0
,
azure_mgmt_resource 23.2.0
Whenever I run a test on this (via Home > AutomationAccount > Process Automation > Runbooks > <RunbookName> > Edit > Edit in Portal > Test Pane
), it constantly returns the following message:
Failed
Traceback (most recent call last):
File "/usr/src/tmp/2e1ee55b-678c-4023-8759-4e290498c63c/runbooks/vm-monitor-logging.py", line 12, in <module>
from azure.mgmtpute import ComputeManagementClient
File "C:\userenv\lib\site-packages\azure\mgmt\compute\__init__.py", line 9, in <module>
from ._compute_management_client import ComputeManagementClient
File "C:\userenv\lib\site-packages\azure\mgmt\compute\_compute_management_client.py", line 18, in <module>
from azure.profiles import KnownProfiles, ProfileDefinition
ModuleNotFoundError: No module named 'azure.profiles'
Correct me if Im wrong, but I believe it's missing the azure.profiles
modules from the ComputeManagementClient
, but I don't see how this is missing when I have imported the ComputeManagementClient
.
This is my MRE:
import time
import smtplib
import json
import base64
import hashlib
import hmac
import requests
import uuid
from datetime import datetime, timezone
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from azure.mgmtpute import ComputeManagementClient
from azure.identity import ClientSecretCredential
from functools import wraps
# Azure subscription
SUBSCRIPTION_ID = 'xxx'
# Azure Resources
resource_group_name = 'xxx'
vm_name = 'xxx'
# Logs Workspace Information
LOGS_WORKSPACE_ID = "xxx"
LOGS_WORKSPACE_KEY = "xxx"
LOGS_API_ENDPOINT_REGION = "xxx"
# Data Collection Endpoint
DATA_COLLECTION_ENDPOINT = f"https://vmstatusdce-o0w0.{LOGS_API_ENDPOINT_REGION}-1.ingest.monitor.azure"
def execution_trace(func) -> str:
"""Collects the name of the function where the issue, error or degradation resides."""
@wraps(func)
def wrapper(*args, **kwargs):
calling_function = f"Source: def {func.__name__}()"
try:
return func(*args, **kwargs, calling_function = calling_function)
except Exception as e:
print(f"There was an error in returning the name of the function. Logs available {e}")
return wrapper
def assign_log_number(func) -> str:
"""Generates a log reference number, which will be sent to Azure Monitor portal."""
@wraps(func)
def wrapper(*args, **kwargs):
assign_log_number = generate_incident_reference_number()
custom_message = None
try:
return func(*args, **kwargs, assign_log_number = assign_log_number)
except TypeError as e:
custom_message = f"There was a TypeError Exception associated with assigning the Azure log number {e}"
except Exception as e:
custom_message = f"There was an error in generating a log reference. Logs available {e}"
if custom_message:
print(custom_message)
return wrapper
# Azure Authentication
@execution_trace
@assign_log_number
def service_principal_authentication(calling_function:str = None, assign_log_number = None) -> ComputeManagementClient | None:
"""Checks the Authentication of Azure's Service Principal"""
custom_message = None
try:
credentials = ClientSecretCredential(
client_id = 'xxx',
tenant_id = 'xxx',
client_secret = 'xxx'
)
return ComputeManagementClient(credentials, SUBSCRIPTION_ID)
#except ClientAuthenticationError as e:
# custom_message = f"Client Authentication Error {e} | {calling_function}"
except Exception as e:
custom_message = f"There was an error in authenticating the Service Principal {e} | {calling_function}."
if custom_message:
print(custom_message)
def generate_incident_reference_number() -> str:
"""Generate an incident reference number to be logged in the Azure monitoring logs, using Python's built-in UUID module. UUID4 is applied."""
"""Generates a universally unique identifier (UUID) to be used as a log number. UUID version 4 is used."""
return f"{uuid.uuid4()}"
def generate_authentication_signature(workspace_id:str, workspace_key:str, body) -> list[str,str]:
"""Generates the signature needed for authenticating the request."""
date = datetime.now(timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT')
content_length = str(len(body))
# Prepare the string to sign
string_to_sign = f"POST\n{content_length}\napplication/json\nx-ms-date:{date}\n/api/logs"
# Generate the signature using HMAC and SHA256
signature = base64.b64encode(
hmac.new(
base64.b64decode(workspace_key),
string_to_sign.encode('utf-8'),
hashlib.sha256
).digest()
).decode('utf-8')
# Prepare the request headers
headers = {
"Content-Type" : "application/json",
"Authorization" : f"SharedKey {workspace_id}:{signature}",
"x-ms-date" : date,
"Log-Type" : "CustomLogs"
}
return headers
@execution_trace
@assign_log_number
def log_to_azure_monitor(new_vm_status:str, calling_function:str = None, assign_log_number:str = None) -> None:
"""Logs the incident data to Azure Monitor via HTTP Data Collector API.
Disgreard Log Number data"""
log_data = [{
"TimeGenerated" : datetime.now(timezone.utc).isoformat(),
"VMName" : vm_name,
"VMStatus" : new_vm_status,
"LogNumber" : assign_log_number,
}]
# Convert log data to JSON format
body = json.dumps(log_data)
# Generate the headers with the signature
headers = generate_authentication_signature(LOGS_WORKSPACE_ID, LOGS_WORKSPACE_KEY, body)
# Azure Monitor HTTP API endpoint
LOGS_API_ENDPOINT = f"https://{LOGS_WORKSPACE_ID}.ods.opinsights.azure/api/logs?api-version=2016-04-01"
# Send the POST request to the Azure Monitor API
custom_message = None
try:
response = requests.post(LOGS_API_ENDPOINT, headers=headers, data=body)
if response.status_code == 200:
custom_message = f"There is a new status to the VM {vm_name}: {new_vm_status}."
else:
custom_message = f"There was an error logging to Azure. Logging error {response.status_code}, Response: {response.text}"
except Exception as e:
custom_message = f"There was an error in the request retrieval. Logs available: {e}"
if custom_message:
print(custom_message)
return
def main() -> None:
previous_vm_status = None
while True:
compute_client = service_principal_authentication()
current_vm_status = get_vm_status(compute_client)
if current_vm_status != previous_vm_status:
log_to_azure_monitor(current_vm_status)
previous_vm_status = current_vm_status
time.sleep(300)
if __name__ == "__main__":
main()
What module depencances or alternations to the scripts would I need to action in order for this to be able to see the azure.profiles
module(s).
Many thanks