Skip to content

agentcore_cli.utils.observability

agentcore_cli.utils.observability

Observability utilities for AgentCore Platform CLI.

This module provides utilities for setting up and managing AWS observability features including CloudWatch Transaction Search for cost-effective tracing.

TransactionSearchManager

Manages AWS CloudWatch Transaction Search configuration.

Source code in agentcore_cli/utils/observability.py
Python
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class TransactionSearchManager:
    """Manages AWS CloudWatch Transaction Search configuration."""

    def __init__(self, region: str):
        """Initialize the Transaction Search manager.

        Args:
            region: AWS region to operate in
        """
        self.region = region

        # Validate AWS credentials first
        if not validate_aws_credentials():
            raise ValueError("AWS credentials not configured")

        # Validate region format
        is_valid, error_msg = validate_region(region)
        if not is_valid:
            raise ValueError(f"Invalid region: {error_msg}")

        # Create AWS clients using the session utility
        session = get_aws_session(region=region)
        self.xray_client = session.client("xray")
        self.logs_client = session.client("logs")
        self.account_id = get_aws_account_id()

    def is_transaction_search_enabled(self) -> tuple[bool, str | None]:
        """Check if Transaction Search is enabled.

        Returns:
            Tuple of (is_enabled, status_message)
        """
        try:
            response = self.xray_client.get_trace_segment_destination()
            destination = response.get("Destination")
            status = response.get("Status")

            if destination == "CloudWatchLogs" and status == "ACTIVE":
                return True, "Transaction Search is enabled and active"
            else:
                return False, f"Transaction Search status: destination={destination}, status={status}"

        except ClientError as e:
            error_code = e.response.get("Error", {}).get("Code", "Unknown")
            if error_code == "ResourceNotFoundException":
                return False, "Transaction Search not configured"
            else:
                return False, f"Error checking Transaction Search: {e}"

    def create_resource_policy(self) -> bool:
        """Create the resource policy for X-Ray to send traces to CloudWatch Logs.

        Returns:
            True if policy was created successfully, False otherwise
        """
        try:
            policy_document = {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Sid": "TransactionSearchXRayAccess",
                        "Effect": "Allow",
                        "Principal": {"Service": "xray.amazonaws.com"},
                        "Action": "logs:PutLogEvents",
                        "Resource": [
                            f"arn:aws:logs:{self.region}:{self.account_id}:log-group:aws/spans:*",
                            f"arn:aws:logs:{self.region}:{self.account_id}:log-group:/aws/application-signals/data:*",
                        ],
                        "Condition": {
                            "ArnLike": {"aws:SourceArn": f"arn:aws:logs:{self.region}:{self.account_id}:*"},
                            "StringEquals": {"aws:SourceAccount": self.account_id},
                        },
                    }
                ],
            }

            self.logs_client.put_resource_policy(
                policyName="AgentCoreTransactionSearchPolicy", policyDocument=json.dumps(policy_document)
            )

            return True

        except ClientError as e:
            click.echo(f"   ❌ Failed to create resource policy: {e}")
            return False

    def configure_trace_destination(self) -> bool:
        """Configure X-Ray to send traces to CloudWatch Logs.

        Returns:
            True if destination was configured successfully, False otherwise
        """
        try:
            self.xray_client.update_trace_segment_destination(Destination="CloudWatchLogs")
            return True

        except ClientError as e:
            click.echo(f"   ❌ Failed to configure trace destination: {e}")
            return False

    def configure_indexing_rule(self, sampling_percentage: float = 1.0) -> bool:
        """Configure the span indexing rule.

        Args:
            sampling_percentage: Percentage of spans to index (default 1.0 for free tier)

        Returns:
            True if indexing rule was configured successfully, False otherwise
        """
        try:
            self.xray_client.update_indexing_rule(
                Name="Default", Rule={"Probabilistic": {"DesiredSamplingPercentage": sampling_percentage}}
            )
            return True

        except ClientError as e:
            click.echo(f"   ❌ Failed to configure indexing rule: {e}")
            return False

    def enable_transaction_search(self, sampling_percentage: float = 1.0) -> bool:
        """Enable Transaction Search with all required configuration.

        Args:
            sampling_percentage: Percentage of spans to index (default 1.0 for free tier)

        Returns:
            True if Transaction Search was enabled successfully, False otherwise
        """
        click.echo("   🔄 Enabling Transaction Search...")

        # Step 1: Create resource policy
        click.echo("   📋 Creating resource policy...")
        if not self.create_resource_policy():
            return False

        # Step 2: Configure trace destination
        click.echo("   đŸŽ¯ Configuring trace destination...")
        if not self.configure_trace_destination():
            return False

        # Step 3: Configure indexing rule
        click.echo(f"   📊 Configuring indexing rule ({sampling_percentage}% sampling)...")
        if not self.configure_indexing_rule(sampling_percentage):
            return False

        # Step 4: Wait a moment for configuration to propagate
        click.echo("   âŗ Waiting for configuration to propagate...")
        time.sleep(5)

        # Step 5: Verify configuration
        is_enabled, status_msg = self.is_transaction_search_enabled()
        if is_enabled:
            click.echo("   ✅ Transaction Search enabled successfully")
            click.echo("   📝 Note: It may take up to 10 minutes for spans to become available for search")
            return True
        else:
            click.echo(f"   ❌ Transaction Search enablement verification failed: {status_msg}")
            return False

__init__(region)

Initialize the Transaction Search manager.

Parameters:

Name Type Description Default
region str

AWS region to operate in

 required
Source code in agentcore_cli/utils/observability.py
Python
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, region: str):
    """Initialize the Transaction Search manager.

    Args:
        region: AWS region to operate in
    """
    self.region = region

    # Validate AWS credentials first
    if not validate_aws_credentials():
        raise ValueError("AWS credentials not configured")

    # Validate region format
    is_valid, error_msg = validate_region(region)
    if not is_valid:
        raise ValueError(f"Invalid region: {error_msg}")

    # Create AWS clients using the session utility
    session = get_aws_session(region=region)
    self.xray_client = session.client("xray")
    self.logs_client = session.client("logs")
    self.account_id = get_aws_account_id()

configure_indexing_rule(sampling_percentage=1.0)

Configure the span indexing rule.

Parameters:

Name Type Description Default
sampling_percentage float

Percentage of spans to index (default 1.0 for free tier)

 1.0

Returns:

Type Description
bool

True if indexing rule was configured successfully, False otherwise
Source code in agentcore_cli/utils/observability.py
Python
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def configure_indexing_rule(self, sampling_percentage: float = 1.0) -> bool:
    """Configure the span indexing rule.

    Args:
        sampling_percentage: Percentage of spans to index (default 1.0 for free tier)

    Returns:
        True if indexing rule was configured successfully, False otherwise
    """
    try:
        self.xray_client.update_indexing_rule(
            Name="Default", Rule={"Probabilistic": {"DesiredSamplingPercentage": sampling_percentage}}
        )
        return True

    except ClientError as e:
        click.echo(f"   ❌ Failed to configure indexing rule: {e}")
        return False

configure_trace_destination()

Configure X-Ray to send traces to CloudWatch Logs.

Returns:

Type Description
bool

True if destination was configured successfully, False otherwise
Source code in agentcore_cli/utils/observability.py
Python
103
104
105
106
107
108
109
110
111
112
113
114
115
def configure_trace_destination(self) -> bool:
    """Configure X-Ray to send traces to CloudWatch Logs.

    Returns:
        True if destination was configured successfully, False otherwise
    """
    try:
        self.xray_client.update_trace_segment_destination(Destination="CloudWatchLogs")
        return True

    except ClientError as e:
        click.echo(f"   ❌ Failed to configure trace destination: {e}")
        return False

create_resource_policy()

Create the resource policy for X-Ray to send traces to CloudWatch Logs.

Returns:

Type Description
bool

True if policy was created successfully, False otherwise
Source code in agentcore_cli/utils/observability.py
Python
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def create_resource_policy(self) -> bool:
    """Create the resource policy for X-Ray to send traces to CloudWatch Logs.

    Returns:
        True if policy was created successfully, False otherwise
    """
    try:
        policy_document = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "TransactionSearchXRayAccess",
                    "Effect": "Allow",
                    "Principal": {"Service": "xray.amazonaws.com"},
                    "Action": "logs:PutLogEvents",
                    "Resource": [
                        f"arn:aws:logs:{self.region}:{self.account_id}:log-group:aws/spans:*",
                        f"arn:aws:logs:{self.region}:{self.account_id}:log-group:/aws/application-signals/data:*",
                    ],
                    "Condition": {
                        "ArnLike": {"aws:SourceArn": f"arn:aws:logs:{self.region}:{self.account_id}:*"},
                        "StringEquals": {"aws:SourceAccount": self.account_id},
                    },
                }
            ],
        }

        self.logs_client.put_resource_policy(
            policyName="AgentCoreTransactionSearchPolicy", policyDocument=json.dumps(policy_document)
        )

        return True

    except ClientError as e:
        click.echo(f"   ❌ Failed to create resource policy: {e}")
        return False

Enable Transaction Search with all required configuration.

Parameters:

Name Type Description Default
sampling_percentage float

Percentage of spans to index (default 1.0 for free tier)

 1.0

Returns:

Type Description
bool

True if Transaction Search was enabled successfully, False otherwise
Source code in agentcore_cli/utils/observability.py
Python
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def enable_transaction_search(self, sampling_percentage: float = 1.0) -> bool:
    """Enable Transaction Search with all required configuration.

    Args:
        sampling_percentage: Percentage of spans to index (default 1.0 for free tier)

    Returns:
        True if Transaction Search was enabled successfully, False otherwise
    """
    click.echo("   🔄 Enabling Transaction Search...")

    # Step 1: Create resource policy
    click.echo("   📋 Creating resource policy...")
    if not self.create_resource_policy():
        return False

    # Step 2: Configure trace destination
    click.echo("   đŸŽ¯ Configuring trace destination...")
    if not self.configure_trace_destination():
        return False

    # Step 3: Configure indexing rule
    click.echo(f"   📊 Configuring indexing rule ({sampling_percentage}% sampling)...")
    if not self.configure_indexing_rule(sampling_percentage):
        return False

    # Step 4: Wait a moment for configuration to propagate
    click.echo("   âŗ Waiting for configuration to propagate...")
    time.sleep(5)

    # Step 5: Verify configuration
    is_enabled, status_msg = self.is_transaction_search_enabled()
    if is_enabled:
        click.echo("   ✅ Transaction Search enabled successfully")
        click.echo("   📝 Note: It may take up to 10 minutes for spans to become available for search")
        return True
    else:
        click.echo(f"   ❌ Transaction Search enablement verification failed: {status_msg}")
        return False

is_transaction_search_enabled()

Check if Transaction Search is enabled.

Returns:

Type Description
tuple[bool, str | None]

Tuple of (is_enabled, status_message)
Source code in agentcore_cli/utils/observability.py
Python
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def is_transaction_search_enabled(self) -> tuple[bool, str | None]:
    """Check if Transaction Search is enabled.

    Returns:
        Tuple of (is_enabled, status_message)
    """
    try:
        response = self.xray_client.get_trace_segment_destination()
        destination = response.get("Destination")
        status = response.get("Status")

        if destination == "CloudWatchLogs" and status == "ACTIVE":
            return True, "Transaction Search is enabled and active"
        else:
            return False, f"Transaction Search status: destination={destination}, status={status}"

    except ClientError as e:
        error_code = e.response.get("Error", {}).get("Code", "Unknown")
        if error_code == "ResourceNotFoundException":
            return False, "Transaction Search not configured"
        else:
            return False, f"Error checking Transaction Search: {e}"

get_transaction_search_status(region='us-west-2')

Get detailed Transaction Search status information.

Parameters:

Name Type Description Default
region str

AWS region to check

 'us-west-2'

Returns:

Type Description
dict[str, str | bool | None]

Dictionary with status information
Source code in agentcore_cli/utils/observability.py
Python
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def get_transaction_search_status(region: str = "us-west-2") -> dict[str, str | bool | None]:
    """Get detailed Transaction Search status information.

    Args:
        region: AWS region to check

    Returns:
        Dictionary with status information
    """
    try:
        manager = TransactionSearchManager(region)
        is_enabled, status_msg = manager.is_transaction_search_enabled()

        return {
            "enabled": is_enabled,
            "status_message": status_msg,
            "region": region,
            "account_id": get_aws_account_id(),
        }
    except Exception as e:
        return {"enabled": False, "status_message": f"Error checking status: {e}", "region": region, "account_id": None}

Validate Transaction Search status and enable if needed.

Parameters:

Name Type Description Default
region str

AWS region to check/configure

 'us-west-2'
interactive bool

Whether to prompt user for confirmation

 True

Returns:

Type Description
bool

True if Transaction Search is enabled (or was enabled successfully), False otherwise
Source code in agentcore_cli/utils/observability.py
Python
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def validate_and_enable_transaction_search(region: str = "us-west-2", interactive: bool = True) -> bool:
    """Validate Transaction Search status and enable if needed.

    Args:
        region: AWS region to check/configure
        interactive: Whether to prompt user for confirmation

    Returns:
        True if Transaction Search is enabled (or was enabled successfully), False otherwise
    """
    try:
        manager = TransactionSearchManager(region)

        # Check current status
        is_enabled, status_msg = manager.is_transaction_search_enabled()

        if is_enabled:
            click.echo("   ✅ Transaction Search is already enabled")
            return True

        click.echo(f"   ℹī¸  Transaction Search status: {status_msg}")

        if interactive:
            enable_search = click.confirm(
                "   Would you like to enable Transaction Search for cost-effective observability?", default=True
            )
            if not enable_search:
                click.echo("   ⏭ī¸  Skipped Transaction Search setup")
                return False

        # Get sampling percentage if interactive
        sampling_percentage = 1.0
        if interactive:
            custom_sampling = click.confirm(
                "   Use custom sampling percentage? (default: 1% - free tier)", default=False
            )
            if custom_sampling:
                sampling_percentage = click.prompt("   Enter sampling percentage (1-100)", default=1.0, type=float)
                if sampling_percentage < 1 or sampling_percentage > 100:
                    click.echo("   ❌ Sampling percentage must be between 1 and 100")
                    sampling_percentage = 1.0

        # Enable Transaction Search
        return manager.enable_transaction_search(sampling_percentage)

    except ValueError as e:
        click.echo(f"   ❌ {e}")
        return False
    except Exception as e:
        click.echo(f"   ❌ Unexpected error: {e}")
        return False