Store access logs by using Python SDK V2 - Object Storage Service

OSS records access logs for each bucket and saves them as log files in a specified bucket. Log files are generated hourly using a fixed naming convention.

Prerequisites

Before you begin, make sure you have:

A source bucket for which you want to enable access logging
A target bucket to store the log files
The required RAM permissions:
Operation Required permission
Enable log storage oss:PutBucketLogging
View log storage configuration oss:GetBucketLogging
Disable log storage oss:DeleteBucketLogging

Operation	Required permission
Enable log storage	`oss:PutBucketLogging`
View log storage configuration	`oss:GetBucketLogging`
Disable log storage	`oss:DeleteBucketLogging`

For information about granting RAM permissions, see Grant custom permissions to a RAM user.

Usage notes

The sample code in this topic uses the China (Hangzhou) region (cn-hangzhou) and a public endpoint. To access OSS from other Alibaba Cloud services in the same region, use an internal endpoint instead. For endpoint details, see OSS regions and endpoints.

Method definitions

All examples use import alibabacloud_oss_v2 as oss.

Enable log storage

put_bucket_logging(request: PutBucketLoggingRequest, **kwargs) → PutBucketLoggingResult

View the log storage configuration

get_bucket_logging(request: GetBucketLoggingRequest, **kwargs) → GetBucketLoggingResult

Disable log storage

delete_bucket_logging(request: DeleteBucketLoggingRequest, **kwargs) → DeleteBucketLoggingResult

Parameters

Parameter	Type	Required	Description
`request`	`PutBucketLoggingRequest`	Yes	Request parameters for enabling log storage. See PutBucketLoggingRequest.
`request`	`GetBucketLoggingRequest`	Yes	Request parameters for viewing the log storage configuration. See GetBucketLoggingRequest.
`request`	`DeleteBucketLoggingRequest`	Yes	Request parameters for disabling log storage. See DeleteBucketLoggingRequest.

Return values

Type	Description
`PutBucketLoggingResult`	Result of enabling log storage. See PutBucketLoggingResult.
`GetBucketLoggingResult`	Result of viewing the log storage configuration. See GetBucketLoggingResult.
`DeleteBucketLoggingResult`	Result of disabling log storage. See DeleteBucketLoggingResult.

For complete method definitions, see put_bucket_logging, get_bucket_logging, and delete_bucket_logging.

Examples

Enable log storage

The core call is client.put_bucket_logging(...). The script accepts --region, --bucket (source bucket), --target_bucket, and an optional --target_prefix.

Key parameters:

Parameter	Required	Description
`bucket`	Yes	The source bucket for which to enable access logging.
`target_bucket`	Yes	The bucket that stores the log files.
`target_prefix`	No	A prefix applied to all log object keys in the target bucket. This parameter can be left empty.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket logging sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the source bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--target_bucket', help='The bucket that stores access logs', required=True)
parser.add_argument('--target_prefix', help='The prefix of the log objects. This parameter can be left blank.', default='')

def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Enable log storage: specify the target bucket and optional prefix
    result = client.put_bucket_logging(oss.PutBucketLoggingRequest(
            bucket=args.bucket,
            bucket_logging_status=oss.BucketLoggingStatus(
                logging_enabled=oss.LoggingEnabled(
                    target_bucket=args.target_bucket,
                    target_prefix=args.target_prefix,
                ),
            ),
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

if __name__ == "__main__":
    main()

View the log storage configuration

The core call is client.get_bucket_logging(...).

Response fields:

Field	Type	Description
`status_code`	int	HTTP status code.
`request_id`	str	Unique request identifier.
`bucket_logging_status.logging_enabled.target_bucket`	str	The target bucket storing access logs. `None` if log storage is disabled.
`bucket_logging_status.logging_enabled.target_prefix`	str	The log object prefix. Empty string if not set.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="get bucket logging sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Retrieve the log storage configuration for the bucket
    result = client.get_bucket_logging(oss.GetBucketLoggingRequest(
            bucket=args.bucket,
    ))

    logging_enabled = (
        result.bucket_logging_status and
        result.bucket_logging_status.logging_enabled
    )
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' target bucket: {logging_enabled.target_bucket if logging_enabled else "Not set"},'
          f' target prefix: {logging_enabled.target_prefix if logging_enabled else "Not set"},'
    )

if __name__ == "__main__":
    main()

Disable log storage

The core call is client.delete_bucket_logging(...). This removes the log storage configuration from the bucket.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="delete bucket logging sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Remove the log storage configuration
    result = client.delete_bucket_logging(oss.DeleteBucketLoggingRequest(
            bucket=args.bucket,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

if __name__ == "__main__":
    main()

Configure user-defined log fields

Call put_user_defined_log_fields_config to customize the user_defined_log_fields field in real-time logs. Specify the HTTP request headers and URL query parameters to record for subsequent request analysis.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put user defined log fields config sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')


def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Configure which HTTP headers and query parameters to record in real-time logs
    result = client.put_user_defined_log_fields_config(
        oss.PutUserDefinedLogFieldsConfigRequest(
            bucket=args.bucket,
            user_defined_log_fields_configuration=oss.UserDefinedLogFieldsConfiguration(
                header_set=oss.LoggingHeaderSet(
                    headers=['header1', 'header2'],
                ),
                param_set=oss.LoggingParamSet(
                    parameters=['parameter1', 'parameter2'],
                ),
            ),
        )
    )

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}')


if __name__ == "__main__":
    main()

Query user-defined log fields

Call get_user_defined_log_fields_config to retrieve the current custom configuration of the user_defined_log_fields field.

import argparse
import alibabacloud_oss_v2 as oss


parser = argparse.ArgumentParser(description="get user defined log fields config sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')


def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Retrieve the custom log field configuration
    result = client.get_user_defined_log_fields_config(
        oss.GetUserDefinedLogFieldsConfigRequest(
            bucket=args.bucket,
        ),
    )

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' header set: {result.user_defined_log_fields_configuration.header_set},'
          f' param set: {result.user_defined_log_fields_configuration.param_set},'
    )


if __name__ == "__main__":
    main()

Delete user-defined log fields

Call delete_user_defined_log_fields_config to remove the custom configuration of the user_defined_log_fields field.

import argparse
import alibabacloud_oss_v2 as oss


parser = argparse.ArgumentParser(description="delete user defined log fields config sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')


def main():
    args = parser.parse_args()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Remove the custom log field configuration
    result = client.delete_user_defined_log_fields_config(
        oss.DeleteUserDefinedLogFieldsConfigRequest(
            bucket=args.bucket,
        ),
    )

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}'
    )


if __name__ == "__main__":
    main()