Python SDK V2 を使用してアクセスログを保存する - Object Storage Service

OSS にアクセスすると、多数のアクセスログが生成されます。ロギング機能を使用して、これらのログを指定されたバケットにログファイルとして保存できます。ログファイルは、固定の命名規則に基づいて 1 時間ごとに生成されます。

注意事項

このトピックのサンプルコードでは、中国 (杭州) リージョン ID cn-hangzhou を例として使用します。デフォルトでは、パブリックエンドポイントが使用されます。同じリージョン内の他の Alibaba Cloud サービスから OSS にアクセスする場合は、内部エンドポイントを使用する必要があります。OSS リージョンとエンドポイントのマッピングの詳細については、「OSS リージョンとエンドポイント」をご参照ください。
ロギングを有効にするには、oss:PutBucketLogging 権限が必要です。ロギング設定を表示するには、oss:GetBucketLogging 権限が必要です。ロギングを無効にするには、oss:DeleteBucketLogging 権限が必要です。詳細については、「RAM ユーザーへのカスタム権限の付与」をご参照ください。

メソッド定義

ロギングの有効化

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

ロギング設定の表示

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

ロギングの無効化

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

リクエストパラメーター

パラメーター	タイプ	説明
request	PutBucketLoggingRequest	リクエストパラメーター。詳細については、「PutBucketLoggingRequest」をご参照ください。
	GetBucketLoggingRequest	リクエストパラメーター。詳細については、「GetBucketLoggingRequest」をご参照ください。
	DeleteBucketLoggingRequest	リクエストパラメーター。詳細については、「DeleteBucketLoggingRequest」をご参照ください。

戻り値

タイプ	説明
PutBucketLoggingResult	戻り値。詳細については、「PutBucketLoggingResult」をご参照ください。
GetBucketLoggingResult	戻り値。詳細については、「GetBucketLoggingResult」をご参照ください。
DeleteBucketLoggingResult	戻り値。詳細については、「DeleteBucketLoggingResult」をご参照ください。

ロギングを有効にするための完全な定義については、「put_bucket_logging」をご参照ください。

ロギング設定を表示するための完全な定義については、「get_bucket_logging」をご参照ください。

ロギングを無効にするための完全な定義については、「delete_bucket_logging」をご参照ください。

例

ロギングの有効化

次のサンプルコードは、ロギングを有効にする方法を示しています。

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 empty.', default='')

def main():
    # コマンドライン引数を解析して、ユーザーが入力した値を取得します。
    args = parser.parse_args()

    # ID 検証のために環境変数からアクセス資格情報情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を使用して設定オブジェクトを作成し、認証プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # ユーザーが指定したコマンドライン引数に基づいて、設定オブジェクトのリージョンプロパティを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、設定オブジェクトのエンドポイントプロパティを更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して OSS クライアントを初期化し、OSS と対話します。
    client = oss.Client(cfg)

    # 指定されたバケットのアクセスロギングを設定するリクエストを送信します。
    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,  # ログオブジェクトのプレフィックス。このパラメーターは空の文字列にすることができます。
                ),
            ),
    ))

    # 操作結果のステータスコードとリクエスト ID を出力して、リクエストステータスを確認します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

# このスクリプトが直接実行されると、main 関数を呼び出してロジックの処理を開始します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。プログラムフローはここから始まります。

ロギング設定の表示

次のサンプルコードは、ロギング設定を表示する方法を示しています。

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()

    # ID 検証のために環境変数からアクセス資格情報情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を使用して設定オブジェクトを作成し、認証プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # ユーザーが指定したコマンドライン引数に基づいて、設定オブジェクトのリージョンプロパティを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、設定オブジェクトのエンドポイントプロパティを更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して OSS クライアントを初期化し、OSS と対話します。
    client = oss.Client(cfg)

    # 指定されたバケットのアクセスロギング設定の詳細を取得するリクエストを送信します。
    result = client.get_bucket_logging(oss.GetBucketLoggingRequest(
            bucket=args.bucket,  # バケットの名前。
    ))

    # 操作結果のステータスコードとリクエスト ID を出力して、リクエストステータスを確認します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' target bucket: {result.bucket_logging_status.logging_enabled.target_bucket if result.bucket_logging_status and result.bucket_logging_status.logging_enabled else "Not set"},'  # アクセスログを保存する宛先バケットの名前。このパラメーターが設定されていない場合は、「Not set」と表示されます。
          f' target prefix: {result.bucket_logging_status.logging_enabled.target_prefix if result.bucket_logging_status and result.bucket_logging_status.logging_enabled else "Not set"},'  # ログオブジェクトのプレフィックス。このパラメーターが設定されていない場合は、「Not set」と表示されます。
    )

# このスクリプトが直接実行されると、main 関数を呼び出してロジックの処理を開始します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。プログラムフローはここから始まります。

ロギングの無効化

次のサンプルコードは、ロギングを無効にする方法を示しています。

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()

    # ID 検証のために環境変数からアクセス資格情報情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を使用して設定オブジェクトを作成し、認証プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # ユーザーが指定したコマンドライン引数に基づいて、設定オブジェクトのリージョンプロパティを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、設定オブジェクトのエンドポイントプロパティを更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して OSS クライアントを初期化し、OSS と対話します。
    client = oss.Client(cfg)

    # 指定されたバケットのアクセスロギング設定を削除するリクエストを送信します。
    result = client.delete_bucket_logging(oss.DeleteBucketLoggingRequest(
            bucket=args.bucket,  # バケットの名前。
    ))

    # 操作結果のステータスコードとリクエスト ID を出力して、リクエストステータスを確認します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

# このスクリプトが直接実行されると、main 関数を呼び出してロジックの処理を開始します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。プログラムフローはここから始まります。

ユーザー定義のログフィールドの設定

PutUserDefinedLogFieldsConfig 操作を呼び出して、バケットのリアルタイムログの user_defined_log_fields フィールドをカスタマイズできます。このフィールドに特定のリクエストヘッダーまたはクエリパラメーターを記録して、後続のリクエスト分析を行うことができます。

import argparse
import alibabacloud_oss_v2 as oss  # Alibaba Cloud OSS SDK をインポートします。

# コマンドライン引数パーサーを作成します。
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()

    # 環境変数を使用してアクセス資格情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    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

    # OSS クライアントを初期化します。
    client = oss.Client(cfg)

    # PutUserDefinedLogFieldsConfig リクエストオブジェクトを構築します。
    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'],  # カスタムログに記録する HTTP ヘッダー。
                ),
                param_set=oss.LoggingParamSet(
                    parameters=['parameter1', 'parameter2'],  # カスタムログに記録する URL パラメーター。
                ),
            ),
        )
    )

    # リクエスト結果のステータスコードとリクエスト ID を出力して、デバッグまたは操作が成功したかどうかを確認します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}')


if __name__ == "__main__":
    main()

ユーザー定義のログフィールドのクエリ

GetUserDefinedLogFieldsConfig 操作を呼び出して、バケットのリアルタイムログの user_defined_log_fields フィールドのカスタム設定を取得できます。

import argparse
import alibabacloud_oss_v2 as oss  # Alibaba Cloud OSS SDK をインポートします。


# コマンドライン引数パーサーを作成します。
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()

    # 環境変数を使用してアクセス資格情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    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

    # OSS クライアントを初期化します。
    client = oss.Client(cfg)

    # GetUserDefinedLogFieldsConfig リクエストオブジェクトを構築します。
    result = client.get_user_defined_log_fields_config(
        oss.GetUserDefinedLogFieldsConfigRequest(
            bucket=args.bucket,  # 宛先バケットの名前を指定します。
        ),
    )

    # リクエスト結果のステータスコード、リクエスト ID、およびカスタムログフィールド設定を出力します。
    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()

ユーザー定義のログフィールドの削除

DeleteUserDefinedLogFieldsConfig 操作を呼び出して、バケットのリアルタイムログの user_defined_log_fields フィールドのカスタム設定を削除できます。

import argparse
import alibabacloud_oss_v2 as oss  # Alibaba Cloud OSS SDK モジュールをインポートします。


# コマンドライン引数パーサーを作成します。
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()

    # 環境変数を使用してアクセス資格情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    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

    # OSS クライアントインスタンスを初期化します。
    client = oss.Client(cfg)

    # DeleteUserDefinedLogFieldsConfig リクエストオブジェクトを構築します。
    result = client.delete_user_defined_log_fields_config(
        oss.DeleteUserDefinedLogFieldsConfigRequest(
            bucket=args.bucket,  # 宛先バケットの名前を指定します。
        ),
    )

    # リクエスト結果のステータスコードとリクエスト ID を出力します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}'
    )


if __name__ == "__main__":
    main()