Use Alibaba Cloud SDK for Python to create a trail to deliver events to Simple Log Service - ActionTrail

This tutorial shows you how to use the Alibaba Cloud SDK for Python to programmatically create an ActionTrail trail that delivers events to a Simple Log Service (SLS) Logstore. This is a common approach for long-term event storage and analysis.

Prerequisites

To follow this tutorial, you need the following:

SLS activated. If you have not used SLS before, log on to the SLS console and follow the prompts to activate the service.
Important
Using SLS for this tutorial incurs costs for data storage, queries, and notifications. For details, see SLS billing overview.
A Python development environment. Python 3.8 or later is recommended. For more information, see Install Python and Build a Python development environment on Windows.
Alibaba Cloud credentials configured as environment variables. The script requires ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET, and ALIBABA_CLOUD_ACCOUNT_ID environment variables for authentication. For instructions, see Create an AccessKey pair and Configure environment variables in Linux, macOS, and Windows.
Sufficient RAM permissions. The RAM user associated with your AccessKey pair must have permissions to manage ActionTrail and SLS. For simplicity, you can attach the AliyunActionTrailFullAccess and AliyunLogFullAccess policies to the RAM user.

Step 1: Set up your environment

Open your terminal or command prompt and install the required SDKs for ActionTrail and SLS.

pip install alibabacloud_actiontrail20200706
pip install alibabacloud_sls20201230
pip install alibabacloud_tea_util

Step 2: Write and run the script

Save the following code as a Python file (such as create_trail.py). This script automates the entire setup process:
1. Creates an SLS project to store audit logs.
2. Creates an ActionTrail trail that delivers events to that SLS project.
3. Starts the trail to begin logging.

Before running, modify the following variables in the constants at the top of the AliyunTrailSetup class:

region_id: The region where you want to create the resources.
sls_project_name: A globally unique name for your SLS project.

trail_name: A unique name for your trail within your account.

import os
import sys
from typing import List

# Import all necessary modules
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models

# Import SLS related modules
from alibabacloud_sls20201230.client import Client as SlsClient
from alibabacloud_sls20201230 import models as sls_models

# Import ActionTrail related modules
from alibabacloud_actiontrail20200706.client import Client as ActionTrailClient
from alibabacloud_actiontrail20200706 import models as actiontrail_models


class AliyunTrailSetup:
    """
    An integrated script to automate the complete setup process for Alibaba Cloud ActionTrail.
    1. Creates an SLS project.
    2. Creates an ActionTrail trail and links it to the SLS project.
    3. Starts the trail to begin logging.
    """

    # --- Configuration Constants ---
    # Define configurable parameters as class variables for easy management and modification.
    REGION_ID = 'cn-hangzhou'
    PROJECT_NAME = 'cloud-trail-project-test'  # Name of the SLS project
    TRAIL_NAME = 'cloud-trail-test'            # Name of the ActionTrail trail

    def __init__(self):
        # Get necessary credentials and account information from environment variables
        self.access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
        self.access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        self.account_id = os.environ.get('ALIBABA_CLOUD_ACCOUNT_ID') # This environment variable must be set

        if not all([self.access_key_id, self.access_key_secret, self.account_id]):
            print("Error: Please ensure the following environment variables are set: 'ALIBABA_CLOUD_ACCESS_KEY_ID', 'ALIBABA_CLOUD_ACCESS_KEY_SECRET', and 'ALIBABA_CLOUD_ACCOUNT_ID'")
            sys.exit(1)

    def create_sls_client(self) -> SlsClient:
        """Create and return an SLS client."""
        config = open_api_models.Config(
            access_key_id=self.access_key_id,
            access_key_secret=self.access_key_secret
        )
        config.endpoint = f'{self.REGION_ID}.log.aliyuncs.com'
        return SlsClient(config)

    def create_actiontrail_client(self) -> ActionTrailClient:
        """Create and return an ActionTrail client."""
        config = open_api_models.Config(
            access_key_id=self.access_key_id,
            access_key_secret=self.access_key_secret
        )
        config.endpoint = f'actiontrail.{self.REGION_ID}.aliyuncs.com'
        return ActionTrailClient(config)

    def run(self) -> None:
        """Execute all setup steps in order."""
        
        # --- Step 1: Create SLS project ---
        print(f"Step 1: Creating SLS project '{self.PROJECT_NAME}'...")
        try:
            sls_client = self.create_sls_client()
            create_project_request = sls_models.CreateProjectRequest(
                project_name=self.PROJECT_NAME,
                description='Project for ActionTrail logs'
            )
            runtime = util_models.RuntimeOptions()
            headers = {}
            sls_client.create_project_with_options(create_project_request, headers, runtime)
            print(f"  [Success] SLS project '{self.PROJECT_NAME}' created successfully.")
        except Exception as error:
            # Handle the case where the project already exists and treat it as a non-fatal condition.
            if hasattr(error, 'data') and error.data.get('body', {}).get('errorCode') == 'ProjectAlreadyExist':
                 print(f"  [Info] SLS project '{self.PROJECT_NAME}' already exists, skipping creation.")
            else:
                print(f"  [Failure] Failed to create SLS project: {error}")
                sys.exit(1) # Terminate the script if the first step fails

        # --- Step 2: Create ActionTrail trail ---
        print(f"\nStep 2: Creating ActionTrail trail '{self.TRAIL_NAME}'...")
        try:
            actiontrail_client = self.create_actiontrail_client()
            
            # Dynamically build the SlsProjectArn to avoid hardcoding
            sls_project_arn = f'acs:log:{self.REGION_ID}:{self.account_id}:project/{self.PROJECT_NAME}'
            print(f"  Using SLS project ARN: {sls_project_arn}")

            create_trail_request = actiontrail_models.CreateTrailRequest(
                name=self.TRAIL_NAME,
                sls_project_arn=sls_project_arn
            )
            runtime = util_models.RuntimeOptions()
            actiontrail_client.create_trail_with_options(create_trail_request, runtime)
            print(f"  [Success] ActionTrail trail '{self.TRAIL_NAME}' created successfully.")
        except Exception as error:
            if hasattr(error, 'data') and error.data.get('body', {}).get('ErrorCode') == 'TrailAlreadyExists':
                print(f"  [Info] ActionTrail trail '{self.TRAIL_NAME}' already exists, skipping creation.")
            else:
                print(f"  [Failure] Failed to create ActionTrail trail: {error}")
                sys.exit(1) # Terminate the script if the second step fails

        # --- Step 3: Start Logging for the ActionTrail trail ---
        print(f"\nStep 3: Starting logging for trail '{self.TRAIL_NAME}'...")
        try:
            # Reuse the already created actiontrail_client
            start_logging_request = actiontrail_models.StartLoggingRequest(
                name=self.TRAIL_NAME
            )
            runtime = util_models.RuntimeOptions()
            actiontrail_client.start_logging_with_options(start_logging_request, runtime)
            print(f"  [Success] Trail '{self.TRAIL_NAME}' has been started.")
        except Exception as error:
            print(f"  [Failure] Failed to start trail: {error}")
            sys.exit(1)

        print("\nAll steps completed successfully!")


if __name__ == '__main__':
    # Instantiate and run the setup process
    setup = AliyunTrailSetup()
    setup.run()

Run the script from your terminal:
```
python create_trail.py
```

A successful run will produce output similar to the following. The script is idempotent, meaning you can run it multiple times without causing errors.

Step 1: Creating SLS project 'cloud-trail-project-test'...
  [Success] SLS project 'cloud-trail-project-test' created successfully.

Step 2: Creating ActionTrail trail 'cloud-trail-test'...
  Using SLS project ARN: acs:log:cn-hangzhou:54728174********:project/cloud-trail-project-test
  [Success] ActionTrail trail 'cloud-trail-test' created successfully.

Step 3: Starting logging for trail 'cloud-trail-test'...
  [Success] Trail 'cloud-trail-test' has been started.

All steps completed successfully!

Step 3: Verify the resources in the console

After running the script, verify that the resources were created correctly.

Verify the trail: Log on to the ActionTrail console. In the left-side navigation pane, click Trails. Confirm that your trail (cloud-trail-test) appears in the list and its Status is Enabled.
Verify the log delivery: Log on to the SLS console. In the Projects list, find your new project (cloud-trail-project-test). Inside the project, you will find a Logstore named actiontrail_cloud-trail-test that is now receiving events.
Note
When you create a trail, ActionTrail automatically creates a Logstore named actiontrail_cloud-trail-test in the project that was created in Step 2.

ActionTrail:Create a trail with the Alibaba Cloud SDK for Python

Prerequisites

Step 1: Set up your environment

Step 2: Write and run the script

Step 3: Verify the resources in the console

References