From 3ab7a6492cde54239c7dfa120d0ad3cfedf778e4 Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Fri, 21 Aug 2026 16:11:50 -0400 Subject: [PATCH 1/6] TON-829: Accept Datadog Operator subscription --- .github/workflows/python-test.yml | 4 + .../accept_operator_subscription.py | 930 ++++++++++++++++++ .../accept_operator_subscription_test.py | 857 ++++++++++++++++ .../attach_integration_permissions_test.py | 8 +- .../datadog_integration_permissions.yaml | 96 +- aws_quickstart/release.sh | 15 +- 6 files changed, 1900 insertions(+), 10 deletions(-) create mode 100644 aws_quickstart/accept_operator_subscription.py create mode 100644 aws_quickstart/accept_operator_subscription_test.py diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index 18382cf5..0638a0ec 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -25,3 +25,7 @@ jobs: run: | cd aws_quickstart python -B -S -m unittest attach_integration_permissions_test.py -v + - name: Run Operator subscription unit tests + run: | + cd aws_quickstart + python -B -S -m unittest accept_operator_subscription_test.py -v diff --git a/aws_quickstart/accept_operator_subscription.py b/aws_quickstart/accept_operator_subscription.py new file mode 100644 index 00000000..442ae116 --- /dev/null +++ b/aws_quickstart/accept_operator_subscription.py @@ -0,0 +1,930 @@ +import http.client +import json +import logging +import time +import urllib.error +import urllib.request +import uuid +from datetime import datetime, timezone +from decimal import Decimal, InvalidOperation + +import boto3 +from botocore.auth import SigV4Auth +from botocore.awsrequest import AWSRequest +from botocore.config import Config +import cfnresponse + + +LOGGER = logging.getLogger() +LOGGER.setLevel(logging.INFO) + +DATADOG_OPERATOR_PRODUCT_ID = "6e852b2a-ecbb-431c-9b63-7de0288f4d00" +EKS_RESOURCE_TYPE = "aws:eks:cluster" +MARKETPLACE_CATALOG = "AWSMarketplace" +MARKETPLACE_REGION = "us-east-1" +AGREEMENT_API_VERSION = "2020-03-01" +AGREEMENT_TARGET_PREFIX = "AWSMPCommerceService_v20200301" +AGREEMENT_SIGNING_NAME = "aws-marketplace" +ENTITLEMENT_ATTEMPTS = 24 +ENTITLEMENT_DELAY_SECONDS = 5 +AWS_READ_TIMEOUT_SECONDS = 20 +SDK_CONNECT_TIMEOUT_SECONDS = 5 +SDK_READ_TIMEOUT_SECONDS = 10 +SDK_REQUEST_BUDGET_SECONDS = 32 +CLOUDFORMATION_RESPONSE_BUFFER_SECONDS = 15 +SIGNED_REQUEST_RETRY_DELAYS_SECONDS = (1, 2) +TRANSIENT_HTTP_STATUS_CODES = frozenset({429, 500, 502, 503, 504}) + +AWS_CONFIG = Config( + retries={"total_max_attempts": 2, "mode": "standard"}, + connect_timeout=SDK_CONNECT_TIMEOUT_SECONDS, + read_timeout=SDK_READ_TIMEOUT_SECONDS, +) + + +class SubscriptionError(Exception): + def __init__( + self, + stage, + reason, + message, + *, + offer_id=None, + agreement_request_id=None, + agreement_id=None, + ): + super().__init__(message) + self.stage = stage + self.reason = reason + self.offer_id = offer_id + self.agreement_request_id = agreement_request_id + self.agreement_id = agreement_id + + +# The Python SDK model does not expose these Agreement operations, so use the +# service's documented AWS JSON protocol with the runtime's SigV4 signer. +class SignedAgreementClient: + def __init__( + self, + credentials, + endpoint, + *, + deadline=None, + opener=urllib.request.urlopen, + ): + self._credentials = credentials.get_frozen_credentials() + self._endpoint = endpoint + self._deadline = deadline + self._opener = opener + + def _request_timeout(self, operation): + if self._deadline is None: + return AWS_READ_TIMEOUT_SECONDS + remaining = self._deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError( + f"{operation} was not attempted because the Lambda deadline was reached" + ) + # urllib applies the timeout separately to connection and response reads. + return min(AWS_READ_TIMEOUT_SECONDS, remaining / 2) + + def _send(self, operation, payload): + timeout = self._request_timeout(operation) + body = json.dumps(payload, separators=(",", ":")).encode() + request = AWSRequest( + method="POST", + url=self._endpoint, + data=body, + headers={ + "Content-Type": "application/x-amz-json-1.0", + "X-Amz-Api-Version": AGREEMENT_API_VERSION, + "X-Amz-Target": f"{AGREEMENT_TARGET_PREFIX}.{operation}", + }, + ) + SigV4Auth( + self._credentials, + AGREEMENT_SIGNING_NAME, + MARKETPLACE_REGION, + ).add_auth(request) + prepared = request.prepare() + http_request = urllib.request.Request( + prepared.url, + data=prepared.body, + headers=dict(prepared.headers), + method=prepared.method, + ) + with self._opener(http_request, timeout=timeout) as response: + return response.read() + + def _wait_before_retry(self, operation, retry_number, delay, error): + if ( + self._deadline is not None + and time.monotonic() + delay >= self._deadline + ): + raise RuntimeError( + f"{error}; the Lambda deadline does not allow another attempt" + ) + _log( + "agreement_api", + "retrying", + "transient_aws_api_error", + marketplace_operation=operation, + marketplace_retry_number=retry_number, + error=str(error), + ) + time.sleep(delay) + + def _call(self, operation, payload): + for attempt in range(len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS) + 1): + try: + response_body = self._send(operation, payload) + except urllib.error.HTTPError as error: + try: + error_body = error.read() + body_read_error = None + except (OSError, http.client.IncompleteRead) as read_error: + error_body = b"" + body_read_error = read_error + finally: + error.close() + message = ( + f"{operation} failed with HTTP {error.code}: " + f"{_aws_error_message(error_body)}" + ) + if body_read_error is not None: + message += f" (error body read failed: {body_read_error})" + if ( + error.code not in TRANSIENT_HTTP_STATUS_CODES + or attempt == len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS) + ): + raise RuntimeError(message) from error + self._wait_before_retry( + operation, + attempt + 1, + SIGNED_REQUEST_RETRY_DELAYS_SECONDS[attempt], + message, + ) + continue + except (OSError, http.client.IncompleteRead) as error: + detail = getattr(error, "reason", error) + message = f"{operation} request failed: {detail}" + if attempt == len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS): + raise RuntimeError(message) from error + self._wait_before_retry( + operation, + attempt + 1, + SIGNED_REQUEST_RETRY_DELAYS_SECONDS[attempt], + message, + ) + continue + + if not response_body: + return {} + try: + return json.loads(response_body) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise RuntimeError(f"{operation} returned invalid JSON") from error + + def create_agreement_request(self, **kwargs): + return self._call("CreateAgreementRequest", kwargs) + + def accept_agreement_request(self, **kwargs): + return self._call("AcceptAgreementRequest", kwargs) + + def get_agreement_entitlements(self, **kwargs): + return self._call("GetAgreementEntitlements", kwargs) + + +def _aws_error_message(body): + try: + payload = json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError): + return body.decode(errors="replace") or "empty response" + error_type = payload.get("__type") or payload.get("code") or "AWS service error" + error_type = error_type.rsplit("#", 1)[-1] + message = payload.get("message") or payload.get("Message") + return f"{error_type}: {message}" if message else error_type + + +def parse_resource_types(raw): + if raw is None: + return [] + items = raw.split(",") if isinstance(raw, str) else list(raw) + return [item.strip() for item in items if item and item.strip()] + + +def _physical_resource_id(event): + return event.get("PhysicalResourceId") or ( + f"{event['StackId']}/{event['LogicalResourceId']}" + ) + + +def _send_response(event, context, status, data): + cfnresponse.send( + event, + context, + status, + responseData=data, + physicalResourceId=_physical_resource_id(event), + ) + + +def _log(stage, result, reason, **fields): + payload = { + "marketplace_stage": stage, + "marketplace_result": result, + "marketplace_reason": reason, + "marketplace_control_plane_region": MARKETPLACE_REGION, + **{key: value for key, value in fields.items() if value is not None}, + } + LOGGER.info( + "Datadog Operator Marketplace subscription event %s", + json.dumps(payload, default=str, sort_keys=True), + ) + + +def _require_sdk_request_budget(deadline, operation): + if ( + deadline is not None + and time.monotonic() + SDK_REQUEST_BUDGET_SECONDS >= deadline + ): + raise RuntimeError( + f"{operation} was not attempted because the Lambda deadline is near" + ) + + +def _pages( + client, + operation, + result_key, + *, + error_stage, + error_message, + deadline=None, + **kwargs, +): + try: + paginator = iter(client.get_paginator(operation).paginate(**kwargs)) + while True: + _require_sdk_request_budget(deadline, operation) + try: + page = next(paginator) + except StopIteration: + return + yield from page.get(result_key, []) + except SubscriptionError: + raise + except Exception as error: + raise SubscriptionError( + error_stage, + "aws_api_error", + f"{error_message}: {error}", + ) from error + + +def _api_call(stage, message, operation, *, deadline=None, **kwargs): + try: + _require_sdk_request_budget( + deadline, + getattr(operation, "__name__", "AWS API call"), + ) + return operation(**kwargs) + except Exception as error: + raise SubscriptionError(stage, "aws_api_error", f"{message}: {error}") from error + + +def find_active_agreement(agreement_search_client, *, deadline=None): + filters = [ + {"name": "PartyType", "values": ["Acceptor"]}, + {"name": "AgreementType", "values": ["PurchaseAgreement"]}, + { + "name": "ResourceIdentifier", + "values": [DATADOG_OPERATOR_PRODUCT_ID], + }, + {"name": "Status", "values": ["ACTIVE"]}, + ] + agreements = list( + _pages( + agreement_search_client, + "search_agreements", + "agreementViewSummaries", + error_stage="agreement_discovery", + error_message="Failed to search for an active Datadog Operator Marketplace agreement", + deadline=deadline, + catalog=MARKETPLACE_CATALOG, + filters=filters, + ) + ) + if not agreements: + return None + if len(agreements) != 1: + raise SubscriptionError( + "agreement_discovery", + "multiple_active_agreements", + "Multiple active Datadog Operator Marketplace agreements were returned", + ) + + agreement_id = agreements[0].get("agreementId") + if not agreement_id: + raise SubscriptionError( + "agreement_discovery", + "invalid_response", + "The active Datadog Operator Marketplace agreement has no identifier", + ) + return agreement_id + + +def _is_available(resource, now): + available_from = resource.get("availableFromTime") + expiration = resource.get("expirationTime") + return (available_from is None or now >= available_from) and ( + expiration is None or now < expiration + ) + + +def _validate_purchase_option(option): + offer_id = option.get("purchaseOptionId") + if not offer_id or option.get("purchaseOptionType") != "OFFER": + raise SubscriptionError( + "offer_discovery", + "invalid_offer", + "The Datadog Operator Marketplace purchase option is not an offer", + offer_id=offer_id, + ) + + entities = option.get("associatedEntities", []) + if len(entities) != 1: + raise SubscriptionError( + "offer_discovery", + "invalid_offer", + "The Datadog Operator Marketplace purchase option has unexpected associated entities", + offer_id=offer_id, + ) + entity = entities[0] + if ( + entity.get("product", {}).get("productId") + != DATADOG_OPERATOR_PRODUCT_ID + or entity.get("offer", {}).get("offerId") != offer_id + ): + raise SubscriptionError( + "offer_discovery", + "invalid_offer", + "The Datadog Operator Marketplace purchase option does not match " + "the expected product and offer", + offer_id=offer_id, + ) + return offer_id + + +def _validate_offer(offer_id, offer): + entities = offer.get("associatedEntities", []) + if ( + offer.get("offerId") != offer_id + or not offer.get("agreementProposalId") + or len(entities) != 1 + or entities[0].get("product", {}).get("productId") + != DATADOG_OPERATOR_PRODUCT_ID + or not offer.get("pricingModel") + ): + raise SubscriptionError( + "offer_discovery", + "invalid_offer", + "The Datadog Operator Marketplace offer details are incomplete " + "or do not match the expected product", + offer_id=offer_id, + ) + + +def find_free_offer(discovery_client, now=None, *, deadline=None): + now = now or datetime.now(timezone.utc) + candidates = _pages( + discovery_client, + "list_purchase_options", + "purchaseOptions", + error_stage="offer_discovery", + error_message="Failed to list Datadog Operator Marketplace purchase options", + deadline=deadline, + filters=[ + { + "filterType": "PRODUCT_ID", + "filterValues": [DATADOG_OPERATOR_PRODUCT_ID], + }, + {"filterType": "PURCHASE_OPTION_TYPE", "filterValues": ["OFFER"]}, + ], + ) + + free_offers = [] + candidate_count = 0 + unavailable_count = 0 + ineligible_count = 0 + for option in candidates: + candidate_count += 1 + offer_id = _validate_purchase_option(option) + if not _is_available(option, now): + unavailable_count += 1 + continue + + offer = _api_call( + "offer_discovery", + f"Failed to get Datadog Operator Marketplace offer {offer_id}", + discovery_client.get_offer, + deadline=deadline, + offerId=offer_id, + ) + _validate_offer(offer_id, offer) + if not _is_available(offer, now): + unavailable_count += 1 + continue + if ( + offer["pricingModel"].get("pricingModelType") != "FREE" + or offer.get("badges") + ): + ineligible_count += 1 + continue + free_offers.append(offer) + + if not free_offers: + raise SubscriptionError( + "offer_discovery", + "no_eligible_free_offer", + "No public free Datadog Operator Marketplace offer was returned " + f"(candidates={candidate_count} unavailable={unavailable_count} " + f"nonfree_or_badged={ineligible_count})", + ) + if len(free_offers) != 1: + raise SubscriptionError( + "offer_discovery", + "multiple_free_offers", + "Multiple public free Datadog Operator Marketplace offers were returned", + ) + return free_offers[0] + + +def requested_terms(discovery_client, offer_id, *, deadline=None): + terms = list( + _pages( + discovery_client, + "get_offer_terms", + "offerTerms", + error_stage="offer_terms", + error_message=f"Failed to get Datadog Operator Marketplace offer {offer_id} terms", + deadline=deadline, + offerId=offer_id, + ) + ) + term_ids = [] + seen = set() + supported = {"legalTerm", "supportTerm"} + for term in terms: + present = [name for name in supported if name in term] + if len(present) != 1 or len(term) != 1: + raise SubscriptionError( + "offer_terms", + "unsupported_terms", + "The free Datadog Operator Marketplace offer contains an unsupported term", + offer_id=offer_id, + ) + term_name = present[0] + if term_name in seen: + raise SubscriptionError( + "offer_terms", + "invalid_terms", + f"The Datadog Operator Marketplace offer contains multiple {term_name} values", + offer_id=offer_id, + ) + term_id = term[term_name].get("id") + if not term_id: + raise SubscriptionError( + "offer_terms", + "invalid_terms", + "The Datadog Operator Marketplace offer contains a term without an identifier", + offer_id=offer_id, + ) + seen.add(term_name) + term_ids.append(term_id) + + if seen != supported: + raise SubscriptionError( + "offer_terms", + "invalid_terms", + "The Datadog Operator Marketplace offer must contain exactly one " + "legal and one support term", + offer_id=offer_id, + ) + return [{"id": term_id} for term_id in sorted(term_ids)] + + +def _require_zero(value, field, *, required=False): + if value is None: + if required: + raise SubscriptionError( + "quote_validation", + "unknown_quote_amount", + f"The free Datadog Operator Marketplace quote has no value for {field}", + ) + return + try: + amount = Decimal(value) + except (InvalidOperation, TypeError, ValueError) as error: + raise SubscriptionError( + "quote_validation", + "unknown_quote_amount", + f"The free Datadog Operator Marketplace quote has an invalid value for {field}", + ) from error + if amount != 0: + raise SubscriptionError( + "quote_validation", + "nonzero_quote", + f"The free Datadog Operator Marketplace quote contains a nonzero charge at {field}", + ) + + +def _validate_taxes(taxes, field): + if not taxes: + return + _require_zero(taxes.get("totalAmount"), f"{field}.totalAmount") + for index, item in enumerate(taxes.get("breakdown", [])): + _require_zero(item.get("amount"), f"{field}.breakdown[{index}].amount") + + +def validate_zero_charge_summary(summary): + if summary is None: + raise SubscriptionError( + "quote_validation", + "unknown_quote_amount", + "The free Datadog Operator Marketplace agreement quote has no charge summary", + ) + + _require_zero( + summary.get("newAgreementValue"), + "newAgreementValue", + required=True, + ) + _require_zero( + summary.get("newAgreementValueAfterTax"), + "newAgreementValueAfterTax", + ) + _validate_taxes(summary.get("estimatedTaxes"), "estimatedTaxes") + for index, charge in enumerate(summary.get("expectedCharges", [])): + _require_zero( + charge.get("amount"), + f"expectedCharges[{index}].amount", + required=True, + ) + _require_zero( + charge.get("amountAfterTax"), + f"expectedCharges[{index}].amountAfterTax", + ) + _validate_taxes( + charge.get("estimatedTaxes"), + f"expectedCharges[{index}].estimatedTaxes", + ) + for index, charge in enumerate(summary.get("itemizedCharges", [])): + _require_zero( + charge.get("incrementalChargeAmount"), + f"itemizedCharges[{index}].incrementalChargeAmount", + required=True, + ) + + +def _client_token(event, proposal_id, terms): + seed = "\0".join( + [ + event["StackId"], + event["LogicalResourceId"], + event["RequestId"], + proposal_id, + *(term["id"] for term in terms), + ] + ) + return str(uuid.uuid5(uuid.NAMESPACE_OID, seed)) + + +def create_and_accept_agreement( + event, + discovery_client, + signed_agreement_client, + agreement_search_client, + *, + deadline=None, +): + offer = find_free_offer(discovery_client, deadline=deadline) + offer_id = offer["offerId"] + terms = requested_terms(discovery_client, offer_id, deadline=deadline) + response = _api_call( + "request_creation", + "Failed to create the Datadog Operator Marketplace agreement request", + signed_agreement_client.create_agreement_request, + agreementProposalIdentifier=offer["agreementProposalId"], + clientToken=_client_token(event, offer["agreementProposalId"], terms), + intent="NEW", + requestedTerms=terms, + ) + try: + validate_zero_charge_summary(response.get("chargeSummary")) + except SubscriptionError as error: + error.offer_id = offer_id + raise + agreement_request_id = response.get("agreementRequestId") + if not agreement_request_id: + raise SubscriptionError( + "request_creation", + "invalid_response", + "The Datadog Operator Marketplace agreement request has no identifier", + offer_id=offer_id, + ) + _log( + "request_creation", + "succeeded", + "free_quote_validated", + marketplace_offer_id=offer_id, + marketplace_agreement_request_id=agreement_request_id, + ) + + try: + accepted = signed_agreement_client.accept_agreement_request( + agreementRequestId=agreement_request_id + ) + except Exception as acceptance_error: + try: + agreement_id = find_active_agreement( + agreement_search_client, + deadline=deadline, + ) + except Exception as recovery_error: + raise SubscriptionError( + "acceptance_recovery", + "recovery_failed", + "Failed to accept the Datadog Operator Marketplace agreement " + f"request and could not determine whether it succeeded: {recovery_error}", + offer_id=offer_id, + agreement_request_id=agreement_request_id, + ) from acceptance_error + if agreement_id: + _log( + "request_acceptance", + "succeeded", + "active_agreement_recovered", + marketplace_offer_id=offer_id, + marketplace_agreement_request_id=agreement_request_id, + marketplace_agreement_id=agreement_id, + ) + return agreement_id, True + raise SubscriptionError( + "request_acceptance", + "aws_api_error", + "Failed to accept the Datadog Operator Marketplace agreement " + f"request: {acceptance_error}", + offer_id=offer_id, + agreement_request_id=agreement_request_id, + ) from acceptance_error + + agreement_id = accepted.get("agreementId") + if not agreement_id: + raise SubscriptionError( + "request_acceptance", + "invalid_response", + "The accepted Datadog Operator Marketplace agreement has no identifier", + offer_id=offer_id, + agreement_request_id=agreement_request_id, + ) + _log( + "request_acceptance", + "succeeded", + "agreement_accepted", + marketplace_offer_id=offer_id, + marketplace_agreement_request_id=agreement_request_id, + marketplace_agreement_id=agreement_id, + ) + return agreement_id, True + + +def entitlement_status(signed_agreement_client, agreement_id): + matches = [] + next_token = None + while True: + request = {"agreementId": agreement_id} + if next_token: + request["nextToken"] = next_token + response = _api_call( + "entitlement", + "Failed to get Datadog Operator Marketplace agreement " + f"{agreement_id} entitlements", + signed_agreement_client.get_agreement_entitlements, + **request, + ) + for entitlement in response.get("agreementEntitlements", []): + if entitlement.get("resource", {}).get("id") == DATADOG_OPERATOR_PRODUCT_ID: + matches.append(entitlement) + next_token = response.get("nextToken") + if not next_token: + break + if len(matches) > 1: + raise SubscriptionError( + "entitlement", + "invalid_response", + "Multiple Datadog Operator Marketplace entitlements were returned", + agreement_id=agreement_id, + ) + return matches[0] if matches else None + + +def wait_for_entitlement( + signed_agreement_client, + agreement_id, + *, + attempts=ENTITLEMENT_ATTEMPTS, + delay=ENTITLEMENT_DELAY_SECONDS, + deadline=None, +): + last_status = None + last_reason = None + for attempt in range(attempts): + if ( + deadline is not None + and time.monotonic() + AWS_READ_TIMEOUT_SECONDS >= deadline + ): + break + entitlement = entitlement_status(signed_agreement_client, agreement_id) + status = entitlement.get("status") if entitlement else None + reason = entitlement.get("statusReasonCode") if entitlement else None + last_status = status + last_reason = reason + if status == "PROVISIONED": + _log( + "entitlement", + "succeeded", + "entitlement_provisioned", + marketplace_agreement_id=agreement_id, + marketplace_entitlement_status=status, + marketplace_entitlement_reason=reason, + ) + return + if status in {"FAILED", "DEPROVISIONED"}: + raise SubscriptionError( + "entitlement", + "entitlement_failed", + f"The Datadog Operator Marketplace entitlement is {status} ({reason})", + agreement_id=agreement_id, + ) + if status not in {None, "PENDING", "SCHEDULED"}: + raise SubscriptionError( + "entitlement", + "unsupported_entitlement_status", + f"The Datadog Operator Marketplace entitlement has unsupported status {status}", + agreement_id=agreement_id, + ) + if attempt + 1 < attempts: + if ( + deadline is not None + and time.monotonic() + delay + AWS_READ_TIMEOUT_SECONDS >= deadline + ): + break + time.sleep(delay) + + raise SubscriptionError( + "entitlement", + "entitlement_timeout", + "Timed out waiting for the Datadog Operator Marketplace entitlement " + f"(status={last_status} reason={last_reason})", + agreement_id=agreement_id, + ) + + +def ensure_subscription(event, *, deadline=None): + try: + session = boto3.Session() + discovery_client = session.client( + "marketplace-discovery", + region_name=MARKETPLACE_REGION, + config=AWS_CONFIG, + ) + agreement_search_client = session.client( + "marketplace-agreement", + region_name=MARKETPLACE_REGION, + config=AWS_CONFIG, + ) + credentials = session.get_credentials() + if credentials is None: + raise RuntimeError("No AWS credentials are available") + signed_agreement_client = SignedAgreementClient( + credentials, + agreement_search_client.meta.endpoint_url, + deadline=deadline, + ) + except Exception as error: + raise SubscriptionError( + "sdk_initialization", + "unsupported_sdk", + "The Lambda runtime AWS SDK could not initialize the required Marketplace " + f"clients (boto3={getattr(boto3, '__version__', 'unknown')}): {error}", + ) from error + _log( + "sdk_initialization", + "succeeded", + "clients_created", + boto3_version=getattr(boto3, "__version__", "unknown"), + ) + + agreement_id = find_active_agreement(agreement_search_client, deadline=deadline) + created = False + if agreement_id: + _log( + "agreement_discovery", + "succeeded", + "active_agreement_reused", + marketplace_agreement_id=agreement_id, + ) + else: + _log( + "agreement_discovery", + "succeeded", + "active_agreement_not_found", + ) + agreement_id, created = create_and_accept_agreement( + event, + discovery_client, + signed_agreement_client, + agreement_search_client, + deadline=deadline, + ) + wait_for_entitlement(signed_agreement_client, agreement_id, deadline=deadline) + return agreement_id, created + + +def handler(event, context): + request_type = event["RequestType"] + properties = event["ResourceProperties"] + account_id = properties.get("AccountId") + partition = properties.get("Partition", "aws") + resource_types = parse_resource_types( + properties.get("InstrumentationResourceTypes") + ) + + if request_type == "Delete": + _log( + "cloudformation_delete", + "succeeded", + "agreement_retained", + account_id=account_id, + ) + _send_response(event, context, cfnresponse.SUCCESS, {"AgreementRetained": True}) + return + + if EKS_RESOURCE_TYPE not in resource_types: + _log( + "resource_selection", + "succeeded", + "eks_not_selected", + account_id=account_id, + ) + _send_response(event, context, cfnresponse.SUCCESS, {"Skipped": True}) + return + + if partition != "aws": + error = SubscriptionError( + "partition_validation", + "unsupported_partition", + "Datadog Operator Marketplace subscription acceptance is supported " + "only in the commercial AWS partition", + ) + _log( + error.stage, + "failed", + error.reason, + account_id=account_id, + error=str(error), + ) + _send_response(event, context, cfnresponse.FAILED, {"Message": str(error)}) + return + + try: + remaining_seconds = context.get_remaining_time_in_millis() / 1000 + deadline = time.monotonic() + max( + 0, + remaining_seconds - CLOUDFORMATION_RESPONSE_BUFFER_SECONDS, + ) + agreement_id, created = ensure_subscription(event, deadline=deadline) + _send_response( + event, + context, + cfnresponse.SUCCESS, + {"AgreementId": agreement_id, "AgreementCreated": created}, + ) + except Exception as error: + stage = getattr(error, "stage", "subscription") + reason = getattr(error, "reason", "aws_api_error") + _log( + stage, + "failed", + reason, + account_id=account_id, + marketplace_offer_id=getattr(error, "offer_id", None), + marketplace_agreement_request_id=getattr( + error, "agreement_request_id", None + ), + marketplace_agreement_id=getattr(error, "agreement_id", None), + error=str(error), + ) + LOGGER.exception("Failed to accept the Datadog Operator Marketplace agreement") + _send_response(event, context, cfnresponse.FAILED, {"Message": str(error)}) diff --git a/aws_quickstart/accept_operator_subscription_test.py b/aws_quickstart/accept_operator_subscription_test.py new file mode 100644 index 00000000..c679c810 --- /dev/null +++ b/aws_quickstart/accept_operator_subscription_test.py @@ -0,0 +1,857 @@ +#!/usr/bin/env python3 + +from datetime import datetime, timedelta, timezone +from io import BytesIO +import http.client +from pathlib import Path +import sys +import unittest +import urllib.error +from unittest.mock import MagicMock, call, patch + + +if "boto3" not in sys.modules: + sys.modules["boto3"] = MagicMock() +if "botocore.config" not in sys.modules: + sys.modules["botocore"] = MagicMock() + sys.modules["botocore.auth"] = MagicMock() + sys.modules["botocore.awsrequest"] = MagicMock() + sys.modules["botocore.config"] = MagicMock() +if "cfnresponse" not in sys.modules: + cfnresponse = MagicMock() + cfnresponse.SUCCESS = "SUCCESS" + cfnresponse.FAILED = "FAILED" + sys.modules["cfnresponse"] = cfnresponse + + +from accept_operator_subscription import ( + SignedAgreementClient, + DATADOG_OPERATOR_PRODUCT_ID, + EKS_RESOURCE_TYPE, + SubscriptionError, + create_and_accept_agreement, + entitlement_status, + ensure_subscription, + find_active_agreement, + find_free_offer, + handler, + parse_resource_types, + requested_terms, + validate_zero_charge_summary, + wait_for_entitlement, +) + + +def paginator_client(**operation_pages): + client = MagicMock() + paginators = {} + for operation, pages in operation_pages.items(): + paginator = MagicMock() + paginator.paginate.return_value = pages + paginators[operation] = paginator + client.get_paginator.side_effect = paginators.__getitem__ + client.paginators = paginators + return client + + +def purchase_option(offer_id="offer-1", **overrides): + value = { + "purchaseOptionId": offer_id, + "purchaseOptionType": "OFFER", + "associatedEntities": [ + { + "product": {"productId": DATADOG_OPERATOR_PRODUCT_ID}, + "offer": {"offerId": offer_id}, + } + ], + } + value.update(overrides) + return value + + +def free_offer(offer_id="offer-1", **overrides): + value = { + "offerId": offer_id, + "agreementProposalId": "ap-proposal1", + "associatedEntities": [ + {"product": {"productId": DATADOG_OPERATOR_PRODUCT_ID}} + ], + "pricingModel": {"pricingModelType": "FREE"}, + "badges": [], + } + value.update(overrides) + return value + + +def event(request_type="Create", resource_types=EKS_RESOURCE_TYPE, partition="aws"): + return { + "RequestType": request_type, + "RequestId": "request-1", + "StackId": "arn:aws:cloudformation:us-east-1:123456789012:stack/test/id", + "LogicalResourceId": "DatadogOperatorSubscriptionFunctionTrigger", + "ResourceProperties": { + "AccountId": "123456789012", + "Partition": partition, + "InstrumentationResourceTypes": resource_types, + }, + } + + +class TestTemplate(unittest.TestCase): + def test_template_embeds_subscription_source(self): + template = Path(__file__).with_name( + "datadog_integration_permissions.yaml" + ).read_text() + + self.assertEqual(template.count(""), 1) + self.assertIn(" IncludeEKS:", template) + for resource in ( + "DatadogOperatorSubscriptionLambdaExecutionRole", + "DatadogOperatorSubscriptionFunction", + "DatadogOperatorSubscriptionFunctionTrigger", + ): + self.assertIn(f" {resource}:\n", template) + self.assertEqual(template.count(" Condition: IncludeEKS"), 3) + + def test_template_grants_only_required_marketplace_actions(self): + template = Path(__file__).with_name( + "datadog_integration_permissions.yaml" + ).read_text() + + actions = ( + "ListPurchaseOptions", + "GetOffer", + "GetOfferTerms", + "SearchAgreements", + "CreateAgreementRequest", + "AcceptAgreementRequest", + "GetAgreementEntitlements", + ) + for action in actions: + self.assertEqual(template.count(f"aws-marketplace:{action}\n"), 1) + self.assertNotIn("aws-marketplace:CancelAgreement", template) + self.assertIn(DATADOG_OPERATOR_PRODUCT_ID, template) + self.assertEqual(template.count(' "Null":'), 3) + + def test_release_embeds_subscription_source(self): + release = Path(__file__).with_name("release.sh").read_text() + + self.assertIn( + "embed_python_source datadog_integration_permissions.yaml " + "accept_operator_subscription.py ACCEPT_OPERATOR_SUBSCRIPTION_SOURCE", + release, + ) + + +class TestResourceTypes(unittest.TestCase): + def test_parse_string_and_list(self): + self.assertEqual( + parse_resource_types("aws:ec2:instance, aws:eks:cluster"), + ["aws:ec2:instance", "aws:eks:cluster"], + ) + self.assertEqual( + parse_resource_types(["aws:eks:cluster", ""]), + ["aws:eks:cluster"], + ) + + +class TestAgreementDiscovery(unittest.TestCase): + def test_returns_active_agreement(self): + client = paginator_client( + search_agreements=[ + {"agreementViewSummaries": [{"agreementId": "agreement-1"}]} + ] + ) + + self.assertEqual(find_active_agreement(client), "agreement-1") + client.paginators["search_agreements"].paginate.assert_called_once_with( + catalog="AWSMarketplace", + filters=[ + {"name": "PartyType", "values": ["Acceptor"]}, + {"name": "AgreementType", "values": ["PurchaseAgreement"]}, + { + "name": "ResourceIdentifier", + "values": [DATADOG_OPERATOR_PRODUCT_ID], + }, + {"name": "Status", "values": ["ACTIVE"]}, + ], + ) + + def test_returns_none_when_no_active_agreement_exists(self): + client = paginator_client( + search_agreements=[{"agreementViewSummaries": []}] + ) + + self.assertIsNone(find_active_agreement(client)) + + def test_rejects_multiple_active_agreements(self): + client = paginator_client( + search_agreements=[ + { + "agreementViewSummaries": [ + {"agreementId": "agreement-1"}, + {"agreementId": "agreement-2"}, + ] + } + ] + ) + + with self.assertRaisesRegex(SubscriptionError, "Multiple active"): + find_active_agreement(client) + + +class TestOfferDiscovery(unittest.TestCase): + def _client(self, options, offers): + client = paginator_client( + list_purchase_options=[{"purchaseOptions": options}] + ) + client.get_offer.side_effect = lambda offerId: offers[offerId] + return client + + def test_selects_only_public_free_offer(self): + client = self._client( + [purchase_option()], + {"offer-1": free_offer()}, + ) + + self.assertEqual(find_free_offer(client)["offerId"], "offer-1") + client.paginators["list_purchase_options"].paginate.assert_called_once_with( + filters=[ + { + "filterType": "PRODUCT_ID", + "filterValues": [DATADOG_OPERATOR_PRODUCT_ID], + }, + { + "filterType": "PURCHASE_OPTION_TYPE", + "filterValues": ["OFFER"], + }, + ] + ) + + def test_rejects_nonfree_and_badged_offers(self): + client = self._client( + [purchase_option("paid"), purchase_option("private")], + { + "paid": free_offer( + "paid", pricingModel={"pricingModelType": "CONTRACT"} + ), + "private": free_offer("private", badges=[{"value": "PRIVATE"}]), + }, + ) + + with self.assertRaisesRegex(SubscriptionError, "No public free"): + find_free_offer(client) + + def test_rejects_multiple_free_offers(self): + client = self._client( + [purchase_option("offer-1"), purchase_option("offer-2")], + { + "offer-1": free_offer("offer-1"), + "offer-2": free_offer("offer-2"), + }, + ) + + with self.assertRaisesRegex(SubscriptionError, "Multiple public free"): + find_free_offer(client) + + def test_skips_unavailable_offer(self): + now = datetime.now(timezone.utc) + client = self._client( + [purchase_option(availableFromTime=now + timedelta(hours=1))], + {"offer-1": free_offer()}, + ) + + with self.assertRaisesRegex(SubscriptionError, "unavailable=1"): + find_free_offer(client, now=now) + client.get_offer.assert_not_called() + + def test_rejects_purchase_option_for_another_product(self): + option = purchase_option() + option["associatedEntities"][0]["product"]["productId"] = "other" + client = self._client([option], {"offer-1": free_offer()}) + + with self.assertRaisesRegex(SubscriptionError, "expected product"): + find_free_offer(client) + + +class TestOfferTerms(unittest.TestCase): + def test_returns_stably_sorted_requested_terms(self): + client = paginator_client( + get_offer_terms=[ + { + "offerTerms": [ + {"supportTerm": {"id": "term-support"}}, + {"legalTerm": {"id": "term-legal"}}, + ] + } + ] + ) + + self.assertEqual( + requested_terms(client, "offer-1"), + [{"id": "term-legal"}, {"id": "term-support"}], + ) + + def test_rejects_missing_or_unsupported_terms(self): + missing = paginator_client( + get_offer_terms=[ + {"offerTerms": [{"legalTerm": {"id": "term-legal"}}]} + ] + ) + unsupported = paginator_client( + get_offer_terms=[ + { + "offerTerms": [ + {"legalTerm": {"id": "term-legal"}}, + {"fixedUpfrontPricingTerm": {"id": "term-price"}}, + ] + } + ] + ) + + with self.assertRaisesRegex(SubscriptionError, "exactly one"): + requested_terms(missing, "offer-1") + with self.assertRaisesRegex(SubscriptionError, "unsupported term"): + requested_terms(unsupported, "offer-1") + + +class TestQuoteValidation(unittest.TestCase): + def test_accepts_only_zero_amounts(self): + validate_zero_charge_summary( + { + "newAgreementValue": "0.00", + "newAgreementValueAfterTax": "0", + "estimatedTaxes": { + "totalAmount": "0", + "breakdown": [{"amount": "0.0"}], + }, + "expectedCharges": [ + { + "amount": "0", + "amountAfterTax": "0", + "estimatedTaxes": {"totalAmount": "0"}, + } + ], + "itemizedCharges": [{"incrementalChargeAmount": "0"}], + } + ) + + def test_rejects_nonzero_and_unknown_amounts(self): + with self.assertRaisesRegex(SubscriptionError, "nonzero charge"): + validate_zero_charge_summary({"newAgreementValue": "0.01"}) + with self.assertRaisesRegex(SubscriptionError, "no charge summary"): + validate_zero_charge_summary(None) + with self.assertRaisesRegex(SubscriptionError, "no value"): + validate_zero_charge_summary({}) + + +class TestSignedAgreementClient(unittest.TestCase): + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_signs_and_sends_json_request(self, mock_request, mock_signer): + credentials = MagicMock() + frozen_credentials = credentials.get_frozen_credentials.return_value + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b'{"agreementRequestId":"request-1"}' + prepared.headers = {"Authorization": "signed"} + prepared.method = "POST" + response = MagicMock() + response.__enter__.return_value.read.return_value = b'{"agreementId":"agreement-1"}' + opener = MagicMock(return_value=response) + + client = SignedAgreementClient(credentials, prepared.url, opener=opener) + + self.assertEqual( + client.accept_agreement_request(agreementRequestId="request-1"), + {"agreementId": "agreement-1"}, + ) + request_kwargs = mock_request.call_args.kwargs + self.assertEqual( + request_kwargs["headers"]["X-Amz-Target"], + "AWSMPCommerceService_v20200301.AcceptAgreementRequest", + ) + self.assertEqual( + request_kwargs["data"], b'{"agreementRequestId":"request-1"}' + ) + mock_signer.assert_called_once_with( + frozen_credentials, + "aws-marketplace", + "us-east-1", + ) + mock_signer.return_value.add_auth.assert_called_once_with( + mock_request.return_value + ) + opener.assert_called_once() + + @patch("accept_operator_subscription.time.monotonic", return_value=10) + def test_splits_remaining_deadline_between_connection_and_read( + self, + _mock_monotonic, + ): + client = SignedAgreementClient( + MagicMock(), + "https://agreement-marketplace.us-east-1.amazonaws.com", + deadline=30, + ) + + self.assertEqual(client._request_timeout("GetAgreementEntitlements"), 10) + + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_retries_transient_error_with_new_signature( + self, + mock_request, + mock_signer, + mock_sleep, + ): + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b"{}" + prepared.headers = {"Authorization": "signed"} + prepared.method = "POST" + transient_error = urllib.error.HTTPError( + prepared.url, + 503, + "Unavailable", + {}, + BytesIO(b'{"__type":"ServiceUnavailable","message":"retry"}'), + ) + response = MagicMock() + response.__enter__.return_value.read.return_value = b'{"agreementId":"agreement-1"}' + opener = MagicMock(side_effect=[transient_error, response]) + client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) + + self.assertEqual( + client.accept_agreement_request(agreementRequestId="request-1"), + {"agreementId": "agreement-1"}, + ) + self.assertEqual(mock_request.call_count, 2) + self.assertEqual(mock_signer.call_count, 2) + mock_sleep.assert_called_once_with(1) + + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_retries_response_read_timeout( + self, + mock_request, + mock_signer, + mock_sleep, + ): + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b"{}" + prepared.headers = {"Authorization": "signed"} + prepared.method = "POST" + timed_out_response = MagicMock() + timed_out_response.__enter__.return_value.read.side_effect = TimeoutError( + "timed out" + ) + successful_response = MagicMock() + successful_response.__enter__.return_value.read.return_value = ( + b'{"agreementId":"agreement-1"}' + ) + opener = MagicMock(side_effect=[timed_out_response, successful_response]) + client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) + + self.assertEqual( + client.accept_agreement_request(agreementRequestId="request-1"), + {"agreementId": "agreement-1"}, + ) + self.assertEqual(mock_request.call_count, 2) + self.assertEqual(mock_signer.call_count, 2) + mock_sleep.assert_called_once_with(1) + + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_retries_incomplete_response_body( + self, + mock_request, + mock_signer, + mock_sleep, + ): + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b"{}" + prepared.headers = {"Authorization": "signed"} + prepared.method = "POST" + incomplete_response = MagicMock() + incomplete_response.__enter__.return_value.read.side_effect = ( + http.client.IncompleteRead(b"partial") + ) + successful_response = MagicMock() + successful_response.__enter__.return_value.read.return_value = ( + b'{"agreementId":"agreement-1"}' + ) + opener = MagicMock(side_effect=[incomplete_response, successful_response]) + client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) + + self.assertEqual( + client.accept_agreement_request(agreementRequestId="request-1"), + {"agreementId": "agreement-1"}, + ) + self.assertEqual(mock_request.call_count, 2) + self.assertEqual(mock_signer.call_count, 2) + mock_sleep.assert_called_once_with(1) + + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_retries_transient_http_error_when_body_read_fails( + self, + mock_request, + mock_signer, + mock_sleep, + ): + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b"{}" + prepared.headers = {"Authorization": "signed"} + prepared.method = "POST" + error_body = MagicMock() + error_body.read.side_effect = TimeoutError("timed out") + transient_error = urllib.error.HTTPError( + prepared.url, + 503, + "Unavailable", + {}, + error_body, + ) + successful_response = MagicMock() + successful_response.__enter__.return_value.read.return_value = ( + b'{"agreementId":"agreement-1"}' + ) + opener = MagicMock(side_effect=[transient_error, successful_response]) + client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) + + self.assertEqual( + client.accept_agreement_request(agreementRequestId="request-1"), + {"agreementId": "agreement-1"}, + ) + self.assertEqual(mock_request.call_count, 2) + self.assertEqual(mock_signer.call_count, 2) + mock_sleep.assert_called_once_with(1) + + @patch("accept_operator_subscription.SigV4Auth") + @patch("accept_operator_subscription.AWSRequest") + def test_includes_aws_error_details(self, mock_request, _mock_signer): + prepared = mock_request.return_value.prepare.return_value + prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" + prepared.body = b"{}" + prepared.headers = {} + prepared.method = "POST" + error = urllib.error.HTTPError( + prepared.url, + 403, + "Forbidden", + {}, + BytesIO(b'{"__type":"AccessDeniedException","message":"denied"}'), + ) + client = SignedAgreementClient( + MagicMock(), + prepared.url, + opener=MagicMock(side_effect=error), + ) + + try: + with self.assertRaisesRegex( + RuntimeError, "HTTP 403: AccessDeniedException: denied" + ): + client.create_agreement_request(intent="NEW", requestedTerms=[]) + finally: + error.close() + + +class TestAgreementCreation(unittest.TestCase): + @patch("accept_operator_subscription.find_free_offer") + @patch("accept_operator_subscription.requested_terms") + def test_creates_validates_and_accepts_agreement(self, mock_terms, mock_offer): + discovery = MagicMock() + agreement = MagicMock() + mock_offer.return_value = free_offer() + mock_terms.return_value = [ + {"id": "term-legal"}, + {"id": "term-support"}, + ] + agreement.create_agreement_request.return_value = { + "agreementRequestId": "request-1", + "chargeSummary": {"newAgreementValue": "0"}, + } + agreement.accept_agreement_request.return_value = { + "agreementId": "agreement-1" + } + + self.assertEqual( + create_and_accept_agreement(event(), discovery, agreement, agreement), + ("agreement-1", True), + ) + create_call = agreement.create_agreement_request.call_args.kwargs + self.assertEqual(create_call["agreementProposalIdentifier"], "ap-proposal1") + self.assertEqual(create_call["intent"], "NEW") + self.assertEqual(create_call["requestedTerms"], mock_terms.return_value) + self.assertEqual(len(create_call["clientToken"]), 36) + agreement.accept_agreement_request.assert_called_once_with( + agreementRequestId="request-1" + ) + + @patch("accept_operator_subscription.find_active_agreement") + @patch("accept_operator_subscription.find_free_offer") + @patch("accept_operator_subscription.requested_terms") + def test_recovers_ambiguous_acceptance( + self, + mock_terms, + mock_offer, + mock_find_active, + ): + mock_offer.return_value = free_offer() + mock_terms.return_value = [{"id": "legal"}, {"id": "support"}] + mock_find_active.return_value = "agreement-1" + agreement = MagicMock() + agreement.create_agreement_request.return_value = { + "agreementRequestId": "request-1", + "chargeSummary": {"newAgreementValue": "0"}, + } + agreement.accept_agreement_request.side_effect = TimeoutError("timed out") + + self.assertEqual( + create_and_accept_agreement(event(), MagicMock(), agreement, agreement), + ("agreement-1", True), + ) + + +class TestEntitlements(unittest.TestCase): + def test_returns_only_operator_entitlement_across_pages(self): + client = MagicMock() + client.get_agreement_entitlements.side_effect = [ + { + "agreementEntitlements": [ + {"resource": {"id": "other"}, "status": "PROVISIONED"} + ], + "nextToken": "next-page", + }, + { + "agreementEntitlements": [ + { + "resource": {"id": DATADOG_OPERATOR_PRODUCT_ID}, + "status": "PENDING", + } + ] + }, + ] + + self.assertEqual( + entitlement_status(client, "agreement-1")["status"], "PENDING" + ) + self.assertEqual( + client.get_agreement_entitlements.call_args_list, + [ + call(agreementId="agreement-1"), + call(agreementId="agreement-1", nextToken="next-page"), + ], + ) + + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.entitlement_status") + def test_waits_until_provisioned(self, mock_status, mock_sleep): + mock_status.side_effect = [ + {"status": "PENDING", "statusReasonCode": "PROVISIONING_IN_PROGRESS"}, + {"status": "PROVISIONED", "statusReasonCode": "AGREEMENT_ACTIVE"}, + ] + + wait_for_entitlement(MagicMock(), "agreement-1", attempts=2, delay=1) + + mock_sleep.assert_called_once_with(1) + + def test_fails_terminal_status_and_timeout(self): + with patch( + "accept_operator_subscription.entitlement_status", + return_value={"status": "FAILED", "statusReasonCode": "PRODUCT_RESTRICTED"}, + ): + with self.assertRaisesRegex(SubscriptionError, "FAILED"): + wait_for_entitlement(MagicMock(), "agreement-1", attempts=1) + with patch( + "accept_operator_subscription.entitlement_status", return_value=None + ): + with self.assertRaisesRegex(SubscriptionError, "Timed out"): + wait_for_entitlement(MagicMock(), "agreement-1", attempts=1) + + @patch("accept_operator_subscription.time.monotonic", return_value=90) + def test_stops_before_deadline_without_starting_request(self, _mock_monotonic): + client = MagicMock() + + with self.assertRaisesRegex(SubscriptionError, "Timed out"): + wait_for_entitlement(client, "agreement-1", deadline=100) + + client.get_agreement_entitlements.assert_not_called() + + +class TestAPIFailureStages(unittest.TestCase): + def assert_stage(self, expected_stage, operation): + with self.assertRaises(SubscriptionError) as raised: + operation() + self.assertEqual(raised.exception.stage, expected_stage) + self.assertEqual(raised.exception.reason, "aws_api_error") + self.assertIn("AccessDenied", str(raised.exception)) + + def test_agreement_discovery_failure(self): + client = MagicMock() + client.get_paginator.side_effect = RuntimeError("AccessDenied") + + self.assert_stage( + "agreement_discovery", lambda: find_active_agreement(client) + ) + + @patch("accept_operator_subscription.time.monotonic", return_value=90) + def test_agreement_discovery_stops_near_deadline(self, _mock_monotonic): + client = paginator_client( + search_agreements=[{"agreementViewSummaries": []}] + ) + + with self.assertRaisesRegex(SubscriptionError, "deadline is near"): + find_active_agreement(client, deadline=100) + + def test_offer_discovery_failure(self): + client = MagicMock() + client.get_paginator.side_effect = RuntimeError("AccessDenied") + + self.assert_stage("offer_discovery", lambda: find_free_offer(client)) + + def test_offer_terms_failure(self): + client = MagicMock() + client.get_paginator.side_effect = RuntimeError("AccessDenied") + + self.assert_stage( + "offer_terms", lambda: requested_terms(client, "offer-1") + ) + + def test_request_creation_failure(self): + agreement = MagicMock() + agreement.create_agreement_request.side_effect = RuntimeError("AccessDenied") + with ( + patch( + "accept_operator_subscription.find_free_offer", + return_value=free_offer(), + ), + patch( + "accept_operator_subscription.requested_terms", + return_value=[{"id": "legal"}, {"id": "support"}], + ), + ): + self.assert_stage( + "request_creation", + lambda: create_and_accept_agreement( + event(), MagicMock(), agreement, agreement + ), + ) + + def test_request_acceptance_failure(self): + agreement = MagicMock() + agreement.create_agreement_request.return_value = { + "agreementRequestId": "request-1", + "chargeSummary": {"newAgreementValue": "0"}, + } + agreement.accept_agreement_request.side_effect = RuntimeError("AccessDenied") + with ( + patch( + "accept_operator_subscription.find_free_offer", + return_value=free_offer(), + ), + patch( + "accept_operator_subscription.requested_terms", + return_value=[{"id": "legal"}, {"id": "support"}], + ), + patch( + "accept_operator_subscription.find_active_agreement", + return_value=None, + ), + ): + self.assert_stage( + "request_acceptance", + lambda: create_and_accept_agreement( + event(), MagicMock(), agreement, agreement + ), + ) + + +class TestClientInitialization(unittest.TestCase): + @patch("accept_operator_subscription.boto3.Session") + def test_reports_runtime_without_marketplace_discovery(self, mock_session): + mock_session.return_value.client.side_effect = RuntimeError("UnknownServiceError") + + with self.assertRaises(SubscriptionError) as raised: + ensure_subscription(event()) + + self.assertEqual(raised.exception.stage, "sdk_initialization") + self.assertEqual(raised.exception.reason, "unsupported_sdk") + self.assertIn("UnknownServiceError", str(raised.exception)) + + +class TestHandler(unittest.TestCase): + def setUp(self): + self.context = MagicMock() + self.context.get_remaining_time_in_millis.return_value = 300_000 + sys.modules["cfnresponse"].send.reset_mock() + + def response(self): + return sys.modules["cfnresponse"].send.call_args + + @patch("accept_operator_subscription.boto3.client") + def test_delete_retains_agreement_without_aws_calls(self, mock_client): + handler(event(request_type="Delete"), self.context) + + mock_client.assert_not_called() + self.assertEqual(self.response().args[2], "SUCCESS") + self.assertEqual( + self.response().kwargs["responseData"], {"AgreementRetained": True} + ) + + @patch("accept_operator_subscription.boto3.client") + def test_skips_when_eks_is_not_selected(self, mock_client): + handler(event(resource_types="aws:ec2:instance"), self.context) + + mock_client.assert_not_called() + self.assertEqual(self.response().args[2], "SUCCESS") + self.assertEqual(self.response().kwargs["responseData"], {"Skipped": True}) + + def test_fails_eks_on_unsupported_partition(self): + handler(event(partition="aws-us-gov"), self.context) + + self.assertEqual(self.response().args[2], "FAILED") + self.assertIn( + "commercial AWS partition", + self.response().kwargs["responseData"]["Message"], + ) + + @patch("accept_operator_subscription.time.monotonic", return_value=100) + @patch("accept_operator_subscription.ensure_subscription") + def test_returns_agreement_details(self, mock_ensure, _mock_monotonic): + mock_ensure.return_value = ("agreement-1", True) + + handler(event(), self.context) + + self.assertEqual(self.response().args[2], "SUCCESS") + self.assertEqual( + self.response().kwargs["responseData"], + {"AgreementId": "agreement-1", "AgreementCreated": True}, + ) + mock_ensure.assert_called_once_with(event(), deadline=385) + + @patch("accept_operator_subscription.ensure_subscription") + def test_reports_actionable_failure(self, mock_ensure): + mock_ensure.side_effect = SubscriptionError( + "offer_discovery", "no_eligible_free_offer", "No free offer" + ) + + handler(event(), self.context) + + self.assertEqual(self.response().args[2], "FAILED") + self.assertEqual( + self.response().kwargs["responseData"], {"Message": "No free offer"} + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/aws_quickstart/attach_integration_permissions_test.py b/aws_quickstart/attach_integration_permissions_test.py index 50a073b7..baa2c463 100644 --- a/aws_quickstart/attach_integration_permissions_test.py +++ b/aws_quickstart/attach_integration_permissions_test.py @@ -60,21 +60,21 @@ def test_cloudformation_template_uses_source_placeholder(self): template = template_path.read_text() self.assertIn( - " Code:\n ZipFile: |\n \n", + " Code:\n ZipFile: |\n \n", template, ) - self.assertEqual(template.count(""), 1) + self.assertEqual(template.count(""), 1) def test_release_embeds_tested_source(self): release_path = Path(__file__).with_name("release.sh") release = release_path.read_text() self.assertIn( - 'cp datadog_agentless_api_call.py attach_integration_permissions.py "${TEMP_DIR}/"', + 'cp datadog_agentless_api_call.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/"', release, ) self.assertIn( - "embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py", + "embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE", release, ) diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 16b1ae43..70081e1f 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -48,6 +48,12 @@ Parameters: an optional add-on to the broader install. The post-setup add-on sets this to "true" because attaching the instrumentation permissions is the stack's only purpose. Updates that replace existing instrumentation permissions still fail atomically so stale policies are not accepted. +Conditions: + IncludeEKS: + Fn::Not: + - Fn::Equals: + - !Join ["", !Split [",aws:eks:cluster,", !Sub ",${InstrumentationResourceTypes},"]] + - !Sub ",${InstrumentationResourceTypes}," Resources: DatadogAttachIntegrationPermissionsLambdaExecutionRole: Type: AWS::IAM::Role @@ -133,7 +139,7 @@ Resources: Timeout: 300 Code: ZipFile: | - + DatadogAttachIntegrationPermissionsFunctionTrigger: Type: Custom::DatadogAttachIntegrationPermissionsFunctionTrigger Properties: @@ -148,3 +154,91 @@ Resources: DatadogSite: !Ref DatadogSite ManageBasePermissions: !Ref ManageBasePermissions FailOnInstrumentationError: !Ref FailOnInstrumentationError + DatadogOperatorSubscriptionLambdaExecutionRole: + Type: AWS::IAM::Role + Condition: IncludeEKS + Properties: + AssumeRolePolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Principal: + Service: + - lambda.amazonaws.com + Action: + - sts:AssumeRole + Path: "/" + ManagedPolicyArns: + - !Sub "arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole" + Policies: + - PolicyName: datadog-operator-marketplace-subscription + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: aws-marketplace:ListPurchaseOptions + Resource: !Sub arn:${AWS::Partition}:aws-marketplace:::catalog/AWSMarketplace/purchaseOption/* + - Effect: Allow + Action: + - aws-marketplace:GetOffer + - aws-marketplace:GetOfferTerms + Resource: !Sub arn:${AWS::Partition}:aws-marketplace:::catalog/AWSMarketplace/offer/* + - Effect: Allow + Action: + - aws-marketplace:CreateAgreementRequest + - aws-marketplace:AcceptAgreementRequest + Resource: "*" + Condition: + "Null": + aws-marketplace:AgreementType: "false" + aws-marketplace:ProductId: "false" + ForAllValues:StringEquals: + aws-marketplace:AgreementType: + - PurchaseAgreement + ForAnyValue:StringEquals: + aws-marketplace:ProductId: + - 6e852b2a-ecbb-431c-9b63-7de0288f4d00 + - Effect: Allow + Action: aws-marketplace:SearchAgreements + Resource: "*" + Condition: + "Null": + aws-marketplace:AgreementType: "false" + ForAllValues:StringEquals: + aws-marketplace:AgreementType: + - PurchaseAgreement + StringEquals: + aws-marketplace:PartyType: Acceptor + - Effect: Allow + Action: aws-marketplace:GetAgreementEntitlements + Resource: "*" + Condition: + "Null": + aws-marketplace:AgreementType: "false" + ForAllValues:StringEquals: + aws-marketplace:AgreementType: + - PurchaseAgreement + DatadogOperatorSubscriptionFunction: + Type: AWS::Lambda::Function + Condition: IncludeEKS + Properties: + Description: Accepts the free Datadog Operator AWS Marketplace agreement. + Role: !GetAtt DatadogOperatorSubscriptionLambdaExecutionRole.Arn + Handler: index.handler + LoggingConfig: + ApplicationLogLevel: INFO + LogFormat: JSON + Runtime: python3.14 + Timeout: 300 + Code: + ZipFile: | + + DatadogOperatorSubscriptionFunctionTrigger: + Type: Custom::DatadogOperatorSubscriptionFunctionTrigger + Condition: IncludeEKS + Properties: + ServiceToken: !GetAtt DatadogOperatorSubscriptionFunction.Arn + SubscriptionSchemaVersion: "1" + AccountId: !Ref AWS::AccountId + Partition: !Sub "${AWS::Partition}" + InstrumentationResourceTypes: !Ref InstrumentationResourceTypes diff --git a/aws_quickstart/release.sh b/aws_quickstart/release.sh index 5456749c..5a242f68 100755 --- a/aws_quickstart/release.sh +++ b/aws_quickstart/release.sh @@ -44,10 +44,14 @@ upload_versions_json() { embed_python_source() { local template="$1" local source="$2" + local placeholder="${3:-ZIPFILE_PLACEHOLDER}" - perl -i -pe ' - BEGIN { $p = do { local $/; } } - /^(\s+)/ && ( + DD_QUICKSTART_SOURCE_PLACEHOLDER="<${placeholder}>" perl -i -pe ' + BEGIN { + $p = do { local $/; }; + $placeholder = $ENV{"DD_QUICKSTART_SOURCE_PLACEHOLDER"}; + } + /^(\s+)\Q$placeholder\E\s*$/ && ( $_ = join("\n", map { $1 . $_ } split(/\n/, $p)) . "\n" ) ' "$template" < "$source" @@ -134,7 +138,7 @@ trap "rm -rf ${TEMP_DIR}" EXIT # Copy all YAML files to temp directory cp *.yaml "${TEMP_DIR}/" -cp datadog_agentless_api_call.py attach_integration_permissions.py "${TEMP_DIR}/" +cp datadog_agentless_api_call.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/" # Change to temp directory for processing cd "${TEMP_DIR}" @@ -161,7 +165,8 @@ for template in main_workflow.yaml main_extended_workflow.yaml main_v2.yaml main fi done -embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py +embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE +embed_python_source datadog_integration_permissions.yaml accept_operator_subscription.py ACCEPT_OPERATOR_SUBSCRIPTION_SOURCE # Process Agentless Scanning templates for template in datadog_agentless_delegate_role.yaml datadog_agentless_scanning.yaml datadog_agentless_delegate_role_snapshot.yaml datadog_integration_autoscaling_policy.yaml datadog_integration_sds_policy.yaml datadog_agentless_delegate_role_stackset.yaml datadog_agentless_saas.yaml; do From 7b045bedecaa44e5ae5a26ecce9d2e4bfb728a73 Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Fri, 21 Aug 2026 16:17:35 -0400 Subject: [PATCH 2/6] Bump Quick Start template to v4.19.0 --- aws_quickstart/CHANGELOG.md | 4 ++++ aws_quickstart/version.txt | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/aws_quickstart/CHANGELOG.md b/aws_quickstart/CHANGELOG.md index f64d8d14..78bf7bbf 100644 --- a/aws_quickstart/CHANGELOG.md +++ b/aws_quickstart/CHANGELOG.md @@ -1,3 +1,7 @@ +# 4.19.0 (August 21, 2026) + +- Accept the free Datadog Operator AWS Marketplace agreement when EKS instrumentation is selected, allowing the managed add-on installation to proceed without Marketplace permissions on the Datadog integration role. + # 4.18.1 (July 31, 2026) - Track every CloudFormation stack using an instrumenter permissions-boundary policy and safely remove the policy after the final owning stack releases it and no IAM entities remain. diff --git a/aws_quickstart/version.txt b/aws_quickstart/version.txt index 347b96f7..b4abc501 100644 --- a/aws_quickstart/version.txt +++ b/aws_quickstart/version.txt @@ -1 +1 @@ -v4.18.1 +v4.19.0 From ee83376da32563507ccc3fa2825ab2cdf092aca2 Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Fri, 21 Aug 2026 17:01:47 -0400 Subject: [PATCH 3/6] Simplify Marketplace agreement client --- .../accept_operator_subscription.py | 253 ++------------ .../accept_operator_subscription_test.py | 312 ++---------------- .../datadog_integration_permissions.yaml | 1 - 3 files changed, 63 insertions(+), 503 deletions(-) diff --git a/aws_quickstart/accept_operator_subscription.py b/aws_quickstart/accept_operator_subscription.py index 442ae116..467181de 100644 --- a/aws_quickstart/accept_operator_subscription.py +++ b/aws_quickstart/accept_operator_subscription.py @@ -1,16 +1,11 @@ -import http.client import json import logging import time -import urllib.error -import urllib.request import uuid from datetime import datetime, timezone from decimal import Decimal, InvalidOperation import boto3 -from botocore.auth import SigV4Auth -from botocore.awsrequest import AWSRequest from botocore.config import Config import cfnresponse @@ -19,12 +14,8 @@ LOGGER.setLevel(logging.INFO) DATADOG_OPERATOR_PRODUCT_ID = "6e852b2a-ecbb-431c-9b63-7de0288f4d00" -EKS_RESOURCE_TYPE = "aws:eks:cluster" MARKETPLACE_CATALOG = "AWSMarketplace" MARKETPLACE_REGION = "us-east-1" -AGREEMENT_API_VERSION = "2020-03-01" -AGREEMENT_TARGET_PREFIX = "AWSMPCommerceService_v20200301" -AGREEMENT_SIGNING_NAME = "aws-marketplace" ENTITLEMENT_ATTEMPTS = 24 ENTITLEMENT_DELAY_SECONDS = 5 AWS_READ_TIMEOUT_SECONDS = 20 @@ -32,8 +23,6 @@ SDK_READ_TIMEOUT_SECONDS = 10 SDK_REQUEST_BUDGET_SECONDS = 32 CLOUDFORMATION_RESPONSE_BUFFER_SECONDS = 15 -SIGNED_REQUEST_RETRY_DELAYS_SECONDS = (1, 2) -TRANSIENT_HTTP_STATUS_CODES = frozenset({429, 500, 502, 503, 504}) AWS_CONFIG = Config( retries={"total_max_attempts": 2, "mode": "standard"}, @@ -61,158 +50,6 @@ def __init__( self.agreement_id = agreement_id -# The Python SDK model does not expose these Agreement operations, so use the -# service's documented AWS JSON protocol with the runtime's SigV4 signer. -class SignedAgreementClient: - def __init__( - self, - credentials, - endpoint, - *, - deadline=None, - opener=urllib.request.urlopen, - ): - self._credentials = credentials.get_frozen_credentials() - self._endpoint = endpoint - self._deadline = deadline - self._opener = opener - - def _request_timeout(self, operation): - if self._deadline is None: - return AWS_READ_TIMEOUT_SECONDS - remaining = self._deadline - time.monotonic() - if remaining <= 0: - raise RuntimeError( - f"{operation} was not attempted because the Lambda deadline was reached" - ) - # urllib applies the timeout separately to connection and response reads. - return min(AWS_READ_TIMEOUT_SECONDS, remaining / 2) - - def _send(self, operation, payload): - timeout = self._request_timeout(operation) - body = json.dumps(payload, separators=(",", ":")).encode() - request = AWSRequest( - method="POST", - url=self._endpoint, - data=body, - headers={ - "Content-Type": "application/x-amz-json-1.0", - "X-Amz-Api-Version": AGREEMENT_API_VERSION, - "X-Amz-Target": f"{AGREEMENT_TARGET_PREFIX}.{operation}", - }, - ) - SigV4Auth( - self._credentials, - AGREEMENT_SIGNING_NAME, - MARKETPLACE_REGION, - ).add_auth(request) - prepared = request.prepare() - http_request = urllib.request.Request( - prepared.url, - data=prepared.body, - headers=dict(prepared.headers), - method=prepared.method, - ) - with self._opener(http_request, timeout=timeout) as response: - return response.read() - - def _wait_before_retry(self, operation, retry_number, delay, error): - if ( - self._deadline is not None - and time.monotonic() + delay >= self._deadline - ): - raise RuntimeError( - f"{error}; the Lambda deadline does not allow another attempt" - ) - _log( - "agreement_api", - "retrying", - "transient_aws_api_error", - marketplace_operation=operation, - marketplace_retry_number=retry_number, - error=str(error), - ) - time.sleep(delay) - - def _call(self, operation, payload): - for attempt in range(len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS) + 1): - try: - response_body = self._send(operation, payload) - except urllib.error.HTTPError as error: - try: - error_body = error.read() - body_read_error = None - except (OSError, http.client.IncompleteRead) as read_error: - error_body = b"" - body_read_error = read_error - finally: - error.close() - message = ( - f"{operation} failed with HTTP {error.code}: " - f"{_aws_error_message(error_body)}" - ) - if body_read_error is not None: - message += f" (error body read failed: {body_read_error})" - if ( - error.code not in TRANSIENT_HTTP_STATUS_CODES - or attempt == len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS) - ): - raise RuntimeError(message) from error - self._wait_before_retry( - operation, - attempt + 1, - SIGNED_REQUEST_RETRY_DELAYS_SECONDS[attempt], - message, - ) - continue - except (OSError, http.client.IncompleteRead) as error: - detail = getattr(error, "reason", error) - message = f"{operation} request failed: {detail}" - if attempt == len(SIGNED_REQUEST_RETRY_DELAYS_SECONDS): - raise RuntimeError(message) from error - self._wait_before_retry( - operation, - attempt + 1, - SIGNED_REQUEST_RETRY_DELAYS_SECONDS[attempt], - message, - ) - continue - - if not response_body: - return {} - try: - return json.loads(response_body) - except (UnicodeDecodeError, json.JSONDecodeError) as error: - raise RuntimeError(f"{operation} returned invalid JSON") from error - - def create_agreement_request(self, **kwargs): - return self._call("CreateAgreementRequest", kwargs) - - def accept_agreement_request(self, **kwargs): - return self._call("AcceptAgreementRequest", kwargs) - - def get_agreement_entitlements(self, **kwargs): - return self._call("GetAgreementEntitlements", kwargs) - - -def _aws_error_message(body): - try: - payload = json.loads(body) - except (UnicodeDecodeError, json.JSONDecodeError): - return body.decode(errors="replace") or "empty response" - error_type = payload.get("__type") or payload.get("code") or "AWS service error" - error_type = error_type.rsplit("#", 1)[-1] - message = payload.get("message") or payload.get("Message") - return f"{error_type}: {message}" if message else error_type - - -def parse_resource_types(raw): - if raw is None: - return [] - items = raw.split(",") if isinstance(raw, str) else list(raw) - return [item.strip() for item in items if item and item.strip()] - - def _physical_resource_id(event): return event.get("PhysicalResourceId") or ( f"{event['StackId']}/{event['LogicalResourceId']}" @@ -293,7 +130,7 @@ def _api_call(stage, message, operation, *, deadline=None, **kwargs): raise SubscriptionError(stage, "aws_api_error", f"{message}: {error}") from error -def find_active_agreement(agreement_search_client, *, deadline=None): +def find_active_agreement(agreement_client, *, deadline=None): filters = [ {"name": "PartyType", "values": ["Acceptor"]}, {"name": "AgreementType", "values": ["PurchaseAgreement"]}, @@ -305,7 +142,7 @@ def find_active_agreement(agreement_search_client, *, deadline=None): ] agreements = list( _pages( - agreement_search_client, + agreement_client, "search_agreements", "agreementViewSummaries", error_stage="agreement_discovery", @@ -603,8 +440,7 @@ def _client_token(event, proposal_id, terms): def create_and_accept_agreement( event, discovery_client, - signed_agreement_client, - agreement_search_client, + agreement_client, *, deadline=None, ): @@ -614,7 +450,8 @@ def create_and_accept_agreement( response = _api_call( "request_creation", "Failed to create the Datadog Operator Marketplace agreement request", - signed_agreement_client.create_agreement_request, + agreement_client.create_agreement_request, + deadline=deadline, agreementProposalIdentifier=offer["agreementProposalId"], clientToken=_client_token(event, offer["agreementProposalId"], terms), intent="NEW", @@ -642,13 +479,14 @@ def create_and_accept_agreement( ) try: - accepted = signed_agreement_client.accept_agreement_request( + _require_sdk_request_budget(deadline, "accept_agreement_request") + accepted = agreement_client.accept_agreement_request( agreementRequestId=agreement_request_id ) except Exception as acceptance_error: try: agreement_id = find_active_agreement( - agreement_search_client, + agreement_client, deadline=deadline, ) except Exception as recovery_error: @@ -699,26 +537,24 @@ def create_and_accept_agreement( return agreement_id, True -def entitlement_status(signed_agreement_client, agreement_id): - matches = [] - next_token = None - while True: - request = {"agreementId": agreement_id} - if next_token: - request["nextToken"] = next_token - response = _api_call( - "entitlement", - "Failed to get Datadog Operator Marketplace agreement " - f"{agreement_id} entitlements", - signed_agreement_client.get_agreement_entitlements, - **request, - ) - for entitlement in response.get("agreementEntitlements", []): - if entitlement.get("resource", {}).get("id") == DATADOG_OPERATOR_PRODUCT_ID: - matches.append(entitlement) - next_token = response.get("nextToken") - if not next_token: - break +def entitlement_status(agreement_client, agreement_id, *, deadline=None): + matches = [ + entitlement + for entitlement in _pages( + agreement_client, + "get_agreement_entitlements", + "agreementEntitlements", + error_stage="entitlement", + error_message=( + "Failed to get Datadog Operator Marketplace agreement " + f"{agreement_id} entitlements" + ), + deadline=deadline, + agreementId=agreement_id, + ) + if entitlement.get("resource", {}).get("id") + == DATADOG_OPERATOR_PRODUCT_ID + ] if len(matches) > 1: raise SubscriptionError( "entitlement", @@ -730,7 +566,7 @@ def entitlement_status(signed_agreement_client, agreement_id): def wait_for_entitlement( - signed_agreement_client, + agreement_client, agreement_id, *, attempts=ENTITLEMENT_ATTEMPTS, @@ -745,7 +581,9 @@ def wait_for_entitlement( and time.monotonic() + AWS_READ_TIMEOUT_SECONDS >= deadline ): break - entitlement = entitlement_status(signed_agreement_client, agreement_id) + entitlement = entitlement_status( + agreement_client, agreement_id, deadline=deadline + ) status = entitlement.get("status") if entitlement else None reason = entitlement.get("statusReasonCode") if entitlement else None last_status = status @@ -799,19 +637,11 @@ def ensure_subscription(event, *, deadline=None): region_name=MARKETPLACE_REGION, config=AWS_CONFIG, ) - agreement_search_client = session.client( + agreement_client = session.client( "marketplace-agreement", region_name=MARKETPLACE_REGION, config=AWS_CONFIG, ) - credentials = session.get_credentials() - if credentials is None: - raise RuntimeError("No AWS credentials are available") - signed_agreement_client = SignedAgreementClient( - credentials, - agreement_search_client.meta.endpoint_url, - deadline=deadline, - ) except Exception as error: raise SubscriptionError( "sdk_initialization", @@ -826,7 +656,7 @@ def ensure_subscription(event, *, deadline=None): boto3_version=getattr(boto3, "__version__", "unknown"), ) - agreement_id = find_active_agreement(agreement_search_client, deadline=deadline) + agreement_id = find_active_agreement(agreement_client, deadline=deadline) created = False if agreement_id: _log( @@ -844,11 +674,10 @@ def ensure_subscription(event, *, deadline=None): agreement_id, created = create_and_accept_agreement( event, discovery_client, - signed_agreement_client, - agreement_search_client, + agreement_client, deadline=deadline, ) - wait_for_entitlement(signed_agreement_client, agreement_id, deadline=deadline) + wait_for_entitlement(agreement_client, agreement_id, deadline=deadline) return agreement_id, created @@ -857,10 +686,6 @@ def handler(event, context): properties = event["ResourceProperties"] account_id = properties.get("AccountId") partition = properties.get("Partition", "aws") - resource_types = parse_resource_types( - properties.get("InstrumentationResourceTypes") - ) - if request_type == "Delete": _log( "cloudformation_delete", @@ -871,16 +696,6 @@ def handler(event, context): _send_response(event, context, cfnresponse.SUCCESS, {"AgreementRetained": True}) return - if EKS_RESOURCE_TYPE not in resource_types: - _log( - "resource_selection", - "succeeded", - "eks_not_selected", - account_id=account_id, - ) - _send_response(event, context, cfnresponse.SUCCESS, {"Skipped": True}) - return - if partition != "aws": error = SubscriptionError( "partition_validation", diff --git a/aws_quickstart/accept_operator_subscription_test.py b/aws_quickstart/accept_operator_subscription_test.py index c679c810..1b466960 100644 --- a/aws_quickstart/accept_operator_subscription_test.py +++ b/aws_quickstart/accept_operator_subscription_test.py @@ -1,21 +1,16 @@ #!/usr/bin/env python3 from datetime import datetime, timedelta, timezone -from io import BytesIO -import http.client from pathlib import Path import sys import unittest -import urllib.error -from unittest.mock import MagicMock, call, patch +from unittest.mock import MagicMock, patch if "boto3" not in sys.modules: sys.modules["boto3"] = MagicMock() if "botocore.config" not in sys.modules: sys.modules["botocore"] = MagicMock() - sys.modules["botocore.auth"] = MagicMock() - sys.modules["botocore.awsrequest"] = MagicMock() sys.modules["botocore.config"] = MagicMock() if "cfnresponse" not in sys.modules: cfnresponse = MagicMock() @@ -25,9 +20,7 @@ from accept_operator_subscription import ( - SignedAgreementClient, DATADOG_OPERATOR_PRODUCT_ID, - EKS_RESOURCE_TYPE, SubscriptionError, create_and_accept_agreement, entitlement_status, @@ -35,7 +28,6 @@ find_active_agreement, find_free_offer, handler, - parse_resource_types, requested_terms, validate_zero_charge_summary, wait_for_entitlement, @@ -83,7 +75,7 @@ def free_offer(offer_id="offer-1", **overrides): return value -def event(request_type="Create", resource_types=EKS_RESOURCE_TYPE, partition="aws"): +def event(request_type="Create", partition="aws"): return { "RequestType": request_type, "RequestId": "request-1", @@ -92,7 +84,6 @@ def event(request_type="Create", resource_types=EKS_RESOURCE_TYPE, partition="aw "ResourceProperties": { "AccountId": "123456789012", "Partition": partition, - "InstrumentationResourceTypes": resource_types, }, } @@ -143,18 +134,6 @@ def test_release_embeds_subscription_source(self): ) -class TestResourceTypes(unittest.TestCase): - def test_parse_string_and_list(self): - self.assertEqual( - parse_resource_types("aws:ec2:instance, aws:eks:cluster"), - ["aws:ec2:instance", "aws:eks:cluster"], - ) - self.assertEqual( - parse_resource_types(["aws:eks:cluster", ""]), - ["aws:eks:cluster"], - ) - - class TestAgreementDiscovery(unittest.TestCase): def test_returns_active_agreement(self): client = paginator_client( @@ -345,226 +324,6 @@ def test_rejects_nonzero_and_unknown_amounts(self): validate_zero_charge_summary({}) -class TestSignedAgreementClient(unittest.TestCase): - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_signs_and_sends_json_request(self, mock_request, mock_signer): - credentials = MagicMock() - frozen_credentials = credentials.get_frozen_credentials.return_value - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b'{"agreementRequestId":"request-1"}' - prepared.headers = {"Authorization": "signed"} - prepared.method = "POST" - response = MagicMock() - response.__enter__.return_value.read.return_value = b'{"agreementId":"agreement-1"}' - opener = MagicMock(return_value=response) - - client = SignedAgreementClient(credentials, prepared.url, opener=opener) - - self.assertEqual( - client.accept_agreement_request(agreementRequestId="request-1"), - {"agreementId": "agreement-1"}, - ) - request_kwargs = mock_request.call_args.kwargs - self.assertEqual( - request_kwargs["headers"]["X-Amz-Target"], - "AWSMPCommerceService_v20200301.AcceptAgreementRequest", - ) - self.assertEqual( - request_kwargs["data"], b'{"agreementRequestId":"request-1"}' - ) - mock_signer.assert_called_once_with( - frozen_credentials, - "aws-marketplace", - "us-east-1", - ) - mock_signer.return_value.add_auth.assert_called_once_with( - mock_request.return_value - ) - opener.assert_called_once() - - @patch("accept_operator_subscription.time.monotonic", return_value=10) - def test_splits_remaining_deadline_between_connection_and_read( - self, - _mock_monotonic, - ): - client = SignedAgreementClient( - MagicMock(), - "https://agreement-marketplace.us-east-1.amazonaws.com", - deadline=30, - ) - - self.assertEqual(client._request_timeout("GetAgreementEntitlements"), 10) - - @patch("accept_operator_subscription.time.sleep") - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_retries_transient_error_with_new_signature( - self, - mock_request, - mock_signer, - mock_sleep, - ): - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b"{}" - prepared.headers = {"Authorization": "signed"} - prepared.method = "POST" - transient_error = urllib.error.HTTPError( - prepared.url, - 503, - "Unavailable", - {}, - BytesIO(b'{"__type":"ServiceUnavailable","message":"retry"}'), - ) - response = MagicMock() - response.__enter__.return_value.read.return_value = b'{"agreementId":"agreement-1"}' - opener = MagicMock(side_effect=[transient_error, response]) - client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) - - self.assertEqual( - client.accept_agreement_request(agreementRequestId="request-1"), - {"agreementId": "agreement-1"}, - ) - self.assertEqual(mock_request.call_count, 2) - self.assertEqual(mock_signer.call_count, 2) - mock_sleep.assert_called_once_with(1) - - @patch("accept_operator_subscription.time.sleep") - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_retries_response_read_timeout( - self, - mock_request, - mock_signer, - mock_sleep, - ): - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b"{}" - prepared.headers = {"Authorization": "signed"} - prepared.method = "POST" - timed_out_response = MagicMock() - timed_out_response.__enter__.return_value.read.side_effect = TimeoutError( - "timed out" - ) - successful_response = MagicMock() - successful_response.__enter__.return_value.read.return_value = ( - b'{"agreementId":"agreement-1"}' - ) - opener = MagicMock(side_effect=[timed_out_response, successful_response]) - client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) - - self.assertEqual( - client.accept_agreement_request(agreementRequestId="request-1"), - {"agreementId": "agreement-1"}, - ) - self.assertEqual(mock_request.call_count, 2) - self.assertEqual(mock_signer.call_count, 2) - mock_sleep.assert_called_once_with(1) - - @patch("accept_operator_subscription.time.sleep") - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_retries_incomplete_response_body( - self, - mock_request, - mock_signer, - mock_sleep, - ): - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b"{}" - prepared.headers = {"Authorization": "signed"} - prepared.method = "POST" - incomplete_response = MagicMock() - incomplete_response.__enter__.return_value.read.side_effect = ( - http.client.IncompleteRead(b"partial") - ) - successful_response = MagicMock() - successful_response.__enter__.return_value.read.return_value = ( - b'{"agreementId":"agreement-1"}' - ) - opener = MagicMock(side_effect=[incomplete_response, successful_response]) - client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) - - self.assertEqual( - client.accept_agreement_request(agreementRequestId="request-1"), - {"agreementId": "agreement-1"}, - ) - self.assertEqual(mock_request.call_count, 2) - self.assertEqual(mock_signer.call_count, 2) - mock_sleep.assert_called_once_with(1) - - @patch("accept_operator_subscription.time.sleep") - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_retries_transient_http_error_when_body_read_fails( - self, - mock_request, - mock_signer, - mock_sleep, - ): - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b"{}" - prepared.headers = {"Authorization": "signed"} - prepared.method = "POST" - error_body = MagicMock() - error_body.read.side_effect = TimeoutError("timed out") - transient_error = urllib.error.HTTPError( - prepared.url, - 503, - "Unavailable", - {}, - error_body, - ) - successful_response = MagicMock() - successful_response.__enter__.return_value.read.return_value = ( - b'{"agreementId":"agreement-1"}' - ) - opener = MagicMock(side_effect=[transient_error, successful_response]) - client = SignedAgreementClient(MagicMock(), prepared.url, opener=opener) - - self.assertEqual( - client.accept_agreement_request(agreementRequestId="request-1"), - {"agreementId": "agreement-1"}, - ) - self.assertEqual(mock_request.call_count, 2) - self.assertEqual(mock_signer.call_count, 2) - mock_sleep.assert_called_once_with(1) - - @patch("accept_operator_subscription.SigV4Auth") - @patch("accept_operator_subscription.AWSRequest") - def test_includes_aws_error_details(self, mock_request, _mock_signer): - prepared = mock_request.return_value.prepare.return_value - prepared.url = "https://agreement-marketplace.us-east-1.amazonaws.com" - prepared.body = b"{}" - prepared.headers = {} - prepared.method = "POST" - error = urllib.error.HTTPError( - prepared.url, - 403, - "Forbidden", - {}, - BytesIO(b'{"__type":"AccessDeniedException","message":"denied"}'), - ) - client = SignedAgreementClient( - MagicMock(), - prepared.url, - opener=MagicMock(side_effect=error), - ) - - try: - with self.assertRaisesRegex( - RuntimeError, "HTTP 403: AccessDeniedException: denied" - ): - client.create_agreement_request(intent="NEW", requestedTerms=[]) - finally: - error.close() - - class TestAgreementCreation(unittest.TestCase): @patch("accept_operator_subscription.find_free_offer") @patch("accept_operator_subscription.requested_terms") @@ -585,7 +344,7 @@ def test_creates_validates_and_accepts_agreement(self, mock_terms, mock_offer): } self.assertEqual( - create_and_accept_agreement(event(), discovery, agreement, agreement), + create_and_accept_agreement(event(), discovery, agreement), ("agreement-1", True), ) create_call = agreement.create_agreement_request.call_args.kwargs @@ -617,41 +376,36 @@ def test_recovers_ambiguous_acceptance( agreement.accept_agreement_request.side_effect = TimeoutError("timed out") self.assertEqual( - create_and_accept_agreement(event(), MagicMock(), agreement, agreement), + create_and_accept_agreement(event(), MagicMock(), agreement), ("agreement-1", True), ) class TestEntitlements(unittest.TestCase): def test_returns_only_operator_entitlement_across_pages(self): - client = MagicMock() - client.get_agreement_entitlements.side_effect = [ - { - "agreementEntitlements": [ - {"resource": {"id": "other"}, "status": "PROVISIONED"} - ], - "nextToken": "next-page", - }, - { - "agreementEntitlements": [ - { - "resource": {"id": DATADOG_OPERATOR_PRODUCT_ID}, - "status": "PENDING", - } - ] - }, - ] + client = paginator_client( + get_agreement_entitlements=[ + { + "agreementEntitlements": [ + {"resource": {"id": "other"}, "status": "PROVISIONED"} + ] + }, + { + "agreementEntitlements": [ + { + "resource": {"id": DATADOG_OPERATOR_PRODUCT_ID}, + "status": "PENDING", + } + ] + }, + ] + ) self.assertEqual( entitlement_status(client, "agreement-1")["status"], "PENDING" ) - self.assertEqual( - client.get_agreement_entitlements.call_args_list, - [ - call(agreementId="agreement-1"), - call(agreementId="agreement-1", nextToken="next-page"), - ], - ) + paginator = client.paginators["get_agreement_entitlements"] + paginator.paginate.assert_called_once_with(agreementId="agreement-1") @patch("accept_operator_subscription.time.sleep") @patch("accept_operator_subscription.entitlement_status") @@ -685,7 +439,7 @@ def test_stops_before_deadline_without_starting_request(self, _mock_monotonic): with self.assertRaisesRegex(SubscriptionError, "Timed out"): wait_for_entitlement(client, "agreement-1", deadline=100) - client.get_agreement_entitlements.assert_not_called() + client.get_paginator.assert_not_called() class TestAPIFailureStages(unittest.TestCase): @@ -743,7 +497,7 @@ def test_request_creation_failure(self): self.assert_stage( "request_creation", lambda: create_and_accept_agreement( - event(), MagicMock(), agreement, agreement + event(), MagicMock(), agreement ), ) @@ -771,7 +525,7 @@ def test_request_acceptance_failure(self): self.assert_stage( "request_acceptance", lambda: create_and_accept_agreement( - event(), MagicMock(), agreement, agreement + event(), MagicMock(), agreement ), ) @@ -798,24 +552,16 @@ def setUp(self): def response(self): return sys.modules["cfnresponse"].send.call_args - @patch("accept_operator_subscription.boto3.client") - def test_delete_retains_agreement_without_aws_calls(self, mock_client): + @patch("accept_operator_subscription.boto3.Session") + def test_delete_retains_agreement_without_aws_calls(self, mock_session): handler(event(request_type="Delete"), self.context) - mock_client.assert_not_called() + mock_session.assert_not_called() self.assertEqual(self.response().args[2], "SUCCESS") self.assertEqual( self.response().kwargs["responseData"], {"AgreementRetained": True} ) - @patch("accept_operator_subscription.boto3.client") - def test_skips_when_eks_is_not_selected(self, mock_client): - handler(event(resource_types="aws:ec2:instance"), self.context) - - mock_client.assert_not_called() - self.assertEqual(self.response().args[2], "SUCCESS") - self.assertEqual(self.response().kwargs["responseData"], {"Skipped": True}) - def test_fails_eks_on_unsupported_partition(self): handler(event(partition="aws-us-gov"), self.context) diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 70081e1f..8976f20d 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -241,4 +241,3 @@ Resources: SubscriptionSchemaVersion: "1" AccountId: !Ref AWS::AccountId Partition: !Sub "${AWS::Partition}" - InstrumentationResourceTypes: !Ref InstrumentationResourceTypes From c9ffa901606f152f4e6fdc132339fc8904084901 Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Fri, 21 Aug 2026 18:14:17 -0400 Subject: [PATCH 4/6] TON-829: Refine Operator subscription handling --- aws_quickstart/CHANGELOG.md | 2 +- .../accept_operator_subscription.py | 40 ++++----- .../accept_operator_subscription_test.py | 82 ++++++++++++++++++- .../datadog_integration_permissions.yaml | 25 ++++-- aws_quickstart/datadog_integration_role.yaml | 4 +- 5 files changed, 118 insertions(+), 35 deletions(-) diff --git a/aws_quickstart/CHANGELOG.md b/aws_quickstart/CHANGELOG.md index 78bf7bbf..2d00d62c 100644 --- a/aws_quickstart/CHANGELOG.md +++ b/aws_quickstart/CHANGELOG.md @@ -1,6 +1,6 @@ # 4.19.0 (August 21, 2026) -- Accept the free Datadog Operator AWS Marketplace agreement when EKS instrumentation is selected, allowing the managed add-on installation to proceed without Marketplace permissions on the Datadog integration role. +- Accept the free Datadog Operator AWS Marketplace agreement in the commercial AWS partition when EKS instrumentation is selected, allowing the managed add-on installation to proceed without Marketplace permissions on the Datadog integration role. GovCloud and China deployments skip automatic agreement acceptance. # 4.18.1 (July 31, 2026) diff --git a/aws_quickstart/accept_operator_subscription.py b/aws_quickstart/accept_operator_subscription.py index 467181de..2f17df89 100644 --- a/aws_quickstart/accept_operator_subscription.py +++ b/aws_quickstart/accept_operator_subscription.py @@ -376,12 +376,17 @@ def _require_zero(value, field, *, required=False): ) -def _validate_taxes(taxes, field): - if not taxes: - return - _require_zero(taxes.get("totalAmount"), f"{field}.totalAmount") - for index, item in enumerate(taxes.get("breakdown", [])): - _require_zero(item.get("amount"), f"{field}.breakdown[{index}].amount") +def _validate_amount_fields(node, path=""): + if isinstance(node, dict): + for key, value in node.items(): + field = f"{path}.{key}" if path else key + if isinstance(value, (dict, list)): + _validate_amount_fields(value, field) + elif key.lower().endswith(("amount", "amountaftertax")): + _require_zero(value, field) + elif isinstance(node, list): + for index, item in enumerate(node): + _validate_amount_fields(item, f"{path}[{index}]") def validate_zero_charge_summary(summary): @@ -401,27 +406,19 @@ def validate_zero_charge_summary(summary): summary.get("newAgreementValueAfterTax"), "newAgreementValueAfterTax", ) - _validate_taxes(summary.get("estimatedTaxes"), "estimatedTaxes") for index, charge in enumerate(summary.get("expectedCharges", [])): _require_zero( charge.get("amount"), f"expectedCharges[{index}].amount", required=True, ) - _require_zero( - charge.get("amountAfterTax"), - f"expectedCharges[{index}].amountAfterTax", - ) - _validate_taxes( - charge.get("estimatedTaxes"), - f"expectedCharges[{index}].estimatedTaxes", - ) for index, charge in enumerate(summary.get("itemizedCharges", [])): _require_zero( charge.get("incrementalChargeAmount"), f"itemizedCharges[{index}].incrementalChargeAmount", required=True, ) + _validate_amount_fields(summary) def _client_token(event, proposal_id, terms): @@ -507,7 +504,7 @@ def create_and_accept_agreement( marketplace_agreement_request_id=agreement_request_id, marketplace_agreement_id=agreement_id, ) - return agreement_id, True + return agreement_id raise SubscriptionError( "request_acceptance", "aws_api_error", @@ -534,7 +531,7 @@ def create_and_accept_agreement( marketplace_agreement_request_id=agreement_request_id, marketplace_agreement_id=agreement_id, ) - return agreement_id, True + return agreement_id def entitlement_status(agreement_client, agreement_id, *, deadline=None): @@ -657,7 +654,6 @@ def ensure_subscription(event, *, deadline=None): ) agreement_id = find_active_agreement(agreement_client, deadline=deadline) - created = False if agreement_id: _log( "agreement_discovery", @@ -671,14 +667,14 @@ def ensure_subscription(event, *, deadline=None): "succeeded", "active_agreement_not_found", ) - agreement_id, created = create_and_accept_agreement( + agreement_id = create_and_accept_agreement( event, discovery_client, agreement_client, deadline=deadline, ) wait_for_entitlement(agreement_client, agreement_id, deadline=deadline) - return agreement_id, created + return agreement_id def handler(event, context): @@ -719,12 +715,12 @@ def handler(event, context): 0, remaining_seconds - CLOUDFORMATION_RESPONSE_BUFFER_SECONDS, ) - agreement_id, created = ensure_subscription(event, deadline=deadline) + agreement_id = ensure_subscription(event, deadline=deadline) _send_response( event, context, cfnresponse.SUCCESS, - {"AgreementId": agreement_id, "AgreementCreated": created}, + {"AgreementId": agreement_id}, ) except Exception as error: stage = getattr(error, "stage", "subscription") diff --git a/aws_quickstart/accept_operator_subscription_test.py b/aws_quickstart/accept_operator_subscription_test.py index 1b466960..fefbcfbb 100644 --- a/aws_quickstart/accept_operator_subscription_test.py +++ b/aws_quickstart/accept_operator_subscription_test.py @@ -96,6 +96,26 @@ def test_template_embeds_subscription_source(self): self.assertEqual(template.count(""), 1) self.assertIn(" IncludeEKS:", template) + self.assertIn( + " InstrumentationResourceTypes:\n Type: CommaDelimitedList", + template, + ) + self.assertEqual( + template.count( + 'NormalizedResourceTypes: !Join [",", ' + "!Ref InstrumentationResourceTypes]" + ), + 2, + ) + self.assertIn( + 'InstrumentationResourceTypes: !Join [",", ' + "!Ref InstrumentationResourceTypes]", + template, + ) + self.assertIn( + " - Fn::Equals:\n - !Ref AWS::Partition\n - aws", + template, + ) for resource in ( "DatadogOperatorSubscriptionLambdaExecutionRole", "DatadogOperatorSubscriptionFunction", @@ -104,6 +124,19 @@ def test_template_embeds_subscription_source(self): self.assertIn(f" {resource}:\n", template) self.assertEqual(template.count(" Condition: IncludeEKS"), 3) + role_template = Path(__file__).with_name( + "datadog_integration_role.yaml" + ).read_text() + self.assertIn( + " InstrumentationResourceTypes:\n Type: CommaDelimitedList", + role_template, + ) + self.assertIn( + 'InstrumentationResourceTypes: !Join [",", ' + "!Ref InstrumentationResourceTypes]", + role_template, + ) + def test_template_grants_only_required_marketplace_actions(self): template = Path(__file__).with_name( "datadog_integration_permissions.yaml" @@ -323,6 +356,47 @@ def test_rejects_nonzero_and_unknown_amounts(self): with self.assertRaisesRegex(SubscriptionError, "no value"): validate_zero_charge_summary({}) + def test_rejects_after_tax_amounts(self): + with self.assertRaisesRegex(SubscriptionError, "amountAfterTax"): + validate_zero_charge_summary( + { + "newAgreementValue": "0", + "expectedCharges": [ + {"amount": "0", "amountAfterTax": "0.01"} + ], + } + ) + + def test_rejects_future_nested_amount_fields(self): + with self.assertRaisesRegex( + SubscriptionError, + r"futureCharges\[0\]\.serviceFeeAmount", + ): + validate_zero_charge_summary( + { + "newAgreementValue": "0", + "futureCharges": [{"serviceFeeAmount": "1"}], + } + ) + + def test_ignores_nonmonetary_value_fields(self): + validate_zero_charge_summary( + { + "newAgreementValue": "0", + "selectorValue": "paid-plan", + "metadata": {"referenceValue": "1"}, + } + ) + + def test_requires_known_charge_amounts(self): + for summary in ( + {"newAgreementValue": "0", "expectedCharges": [{}]}, + {"newAgreementValue": "0", "itemizedCharges": [{}]}, + ): + with self.subTest(summary=summary): + with self.assertRaisesRegex(SubscriptionError, "no value"): + validate_zero_charge_summary(summary) + class TestAgreementCreation(unittest.TestCase): @patch("accept_operator_subscription.find_free_offer") @@ -345,7 +419,7 @@ def test_creates_validates_and_accepts_agreement(self, mock_terms, mock_offer): self.assertEqual( create_and_accept_agreement(event(), discovery, agreement), - ("agreement-1", True), + "agreement-1", ) create_call = agreement.create_agreement_request.call_args.kwargs self.assertEqual(create_call["agreementProposalIdentifier"], "ap-proposal1") @@ -377,7 +451,7 @@ def test_recovers_ambiguous_acceptance( self.assertEqual( create_and_accept_agreement(event(), MagicMock(), agreement), - ("agreement-1", True), + "agreement-1", ) @@ -574,14 +648,14 @@ def test_fails_eks_on_unsupported_partition(self): @patch("accept_operator_subscription.time.monotonic", return_value=100) @patch("accept_operator_subscription.ensure_subscription") def test_returns_agreement_details(self, mock_ensure, _mock_monotonic): - mock_ensure.return_value = ("agreement-1", True) + mock_ensure.return_value = "agreement-1" handler(event(), self.context) self.assertEqual(self.response().args[2], "SUCCESS") self.assertEqual( self.response().kwargs["responseData"], - {"AgreementId": "agreement-1", "AgreementCreated": True}, + {"AgreementId": "agreement-1"}, ) mock_ensure.assert_called_once_with(event(), deadline=385) diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 8976f20d..51585169 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -14,12 +14,13 @@ Parameters: Description: >- Set this value to "true" to add permissions for Datadog to collect resource configuration data. InstrumentationResourceTypes: - Type: String + Type: CommaDelimitedList Default: "" Description: >- Comma-separated list of AWS resource types (UDM form, e.g. aws:ec2:instance, aws:ecs:cluster, aws:eks:cluster) that the Datadog integration role should be granted the IAM permissions - required to instrument with the Datadog Agent. Leave blank to skip. + required to instrument with the Datadog Agent. In the commercial AWS partition, selecting + EKS also accepts the free Datadog Operator AWS Marketplace agreement. Leave blank to skip. DatadogSite: Type: String Default: "datadoghq.com" @@ -50,10 +51,22 @@ Parameters: instrumentation permissions still fail atomically so stale policies are not accepted. Conditions: IncludeEKS: - Fn::Not: + Fn::And: - Fn::Equals: - - !Join ["", !Split [",aws:eks:cluster,", !Sub ",${InstrumentationResourceTypes},"]] - - !Sub ",${InstrumentationResourceTypes}," + - !Ref AWS::Partition + - aws + - Fn::Not: + - Fn::Equals: + - !Join + - "" + - !Split + - ",aws:eks:cluster," + - !Sub + - ",${NormalizedResourceTypes}," + - NormalizedResourceTypes: !Join [",", !Ref InstrumentationResourceTypes] + - !Sub + - ",${NormalizedResourceTypes}," + - NormalizedResourceTypes: !Join [",", !Ref InstrumentationResourceTypes] Resources: DatadogAttachIntegrationPermissionsLambdaExecutionRole: Type: AWS::IAM::Role @@ -150,7 +163,7 @@ Resources: AccountId: !Ref AWS::AccountId Partition: !Sub "${AWS::Partition}" ResourceCollectionPermissions: !Ref ResourceCollectionPermissions - InstrumentationResourceTypes: !Ref InstrumentationResourceTypes + InstrumentationResourceTypes: !Join [",", !Ref InstrumentationResourceTypes] DatadogSite: !Ref DatadogSite ManageBasePermissions: !Ref ManageBasePermissions FailOnInstrumentationError: !Ref FailOnInstrumentationError diff --git a/aws_quickstart/datadog_integration_role.yaml b/aws_quickstart/datadog_integration_role.yaml index 0a3cf14e..f3285ed0 100644 --- a/aws_quickstart/datadog_integration_role.yaml +++ b/aws_quickstart/datadog_integration_role.yaml @@ -19,7 +19,7 @@ Parameters: Description: >- Set this value to "true" to add permissions for Datadog to collect resource configuration data. InstrumentationResourceTypes: - Type: String + Type: CommaDelimitedList Default: "" Description: >- Comma-separated list of AWS resource types (UDM form, e.g. aws:ec2:instance, aws:ecs:cluster, @@ -85,7 +85,7 @@ Resources: Parameters: IAMRoleName: !Ref IAMRoleName ResourceCollectionPermissions: !Ref ResourceCollectionPermissions - InstrumentationResourceTypes: !Ref InstrumentationResourceTypes + InstrumentationResourceTypes: !Join [",", !Ref InstrumentationResourceTypes] DatadogSite: !Ref DatadogSite ManageBasePermissions: true Metadata: From 45a450de370203fda69bc60b08c00e850c7b18ee Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Fri, 21 Aug 2026 18:42:38 -0400 Subject: [PATCH 5/6] TON-829: Share CloudFormation response helpers --- .github/workflows/python-test.yml | 4 + .../accept_operator_subscription.py | 45 ++++++----- .../accept_operator_subscription_test.py | 2 +- .../attach_integration_permissions.py | 30 +++----- .../attach_integration_permissions_test.py | 4 +- aws_quickstart/cfn_common.py | 14 ++++ aws_quickstart/cfn_common_test.py | 76 +++++++++++++++++++ aws_quickstart/release.sh | 23 +++++- 8 files changed, 154 insertions(+), 44 deletions(-) create mode 100644 aws_quickstart/cfn_common.py create mode 100644 aws_quickstart/cfn_common_test.py diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index 0638a0ec..e043a359 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -25,6 +25,10 @@ jobs: run: | cd aws_quickstart python -B -S -m unittest attach_integration_permissions_test.py -v + - name: Run shared CloudFormation helper unit tests + run: | + cd aws_quickstart + python -B -S -m unittest cfn_common_test.py -v - name: Run Operator subscription unit tests run: | cd aws_quickstart diff --git a/aws_quickstart/accept_operator_subscription.py b/aws_quickstart/accept_operator_subscription.py index 2f17df89..1ce629b8 100644 --- a/aws_quickstart/accept_operator_subscription.py +++ b/aws_quickstart/accept_operator_subscription.py @@ -9,6 +9,8 @@ from botocore.config import Config import cfnresponse +from cfn_common import send_cfn_response + LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) @@ -50,22 +52,6 @@ def __init__( self.agreement_id = agreement_id -def _physical_resource_id(event): - return event.get("PhysicalResourceId") or ( - f"{event['StackId']}/{event['LogicalResourceId']}" - ) - - -def _send_response(event, context, status, data): - cfnresponse.send( - event, - context, - status, - responseData=data, - physicalResourceId=_physical_resource_id(event), - ) - - def _log(stage, result, reason, **fields): payload = { "marketplace_stage": stage, @@ -689,7 +675,13 @@ def handler(event, context): "agreement_retained", account_id=account_id, ) - _send_response(event, context, cfnresponse.SUCCESS, {"AgreementRetained": True}) + send_cfn_response( + cfnresponse, + event, + context, + cfnresponse.SUCCESS, + {"AgreementRetained": True}, + ) return if partition != "aws": @@ -706,7 +698,13 @@ def handler(event, context): account_id=account_id, error=str(error), ) - _send_response(event, context, cfnresponse.FAILED, {"Message": str(error)}) + send_cfn_response( + cfnresponse, + event, + context, + cfnresponse.FAILED, + {"Message": str(error)}, + ) return try: @@ -716,7 +714,8 @@ def handler(event, context): remaining_seconds - CLOUDFORMATION_RESPONSE_BUFFER_SECONDS, ) agreement_id = ensure_subscription(event, deadline=deadline) - _send_response( + send_cfn_response( + cfnresponse, event, context, cfnresponse.SUCCESS, @@ -738,4 +737,10 @@ def handler(event, context): error=str(error), ) LOGGER.exception("Failed to accept the Datadog Operator Marketplace agreement") - _send_response(event, context, cfnresponse.FAILED, {"Message": str(error)}) + send_cfn_response( + cfnresponse, + event, + context, + cfnresponse.FAILED, + {"Message": str(error)}, + ) diff --git a/aws_quickstart/accept_operator_subscription_test.py b/aws_quickstart/accept_operator_subscription_test.py index fefbcfbb..d0be9dba 100644 --- a/aws_quickstart/accept_operator_subscription_test.py +++ b/aws_quickstart/accept_operator_subscription_test.py @@ -161,7 +161,7 @@ def test_release_embeds_subscription_source(self): release = Path(__file__).with_name("release.sh").read_text() self.assertIn( - "embed_python_source datadog_integration_permissions.yaml " + "embed_python_source_with_common datadog_integration_permissions.yaml " "accept_operator_subscription.py ACCEPT_OPERATOR_SUBSCRIPTION_SOURCE", release, ) diff --git a/aws_quickstart/attach_integration_permissions.py b/aws_quickstart/attach_integration_permissions.py index 784551fe..39be6110 100644 --- a/aws_quickstart/attach_integration_permissions.py +++ b/aws_quickstart/attach_integration_permissions.py @@ -10,6 +10,8 @@ import cfnresponse import boto3 +from cfn_common import send_cfn_response + LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) API_CALL_SOURCE_HEADER_VALUE = "cfn-quickstart" @@ -142,20 +144,6 @@ def _should_update_instrumentation_permissions(event): ) -def _physical_resource_id(event): - return event.get("PhysicalResourceId") or f"{event['StackId']}/{event['LogicalResourceId']}" - - -def _send_cfn_response(event, context, status, response_data): - cfnresponse.send( - event, - context, - status, - responseData=response_data, - physicalResourceId=_physical_resource_id(event), - ) - - def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( [("resource_type", t) for t in resource_types] @@ -1209,10 +1197,14 @@ def handle_delete(event, context): response_data = {} if preserved_boundaries: response_data["PreservedPermissionsBoundaries"] = preserved_boundaries - _send_cfn_response(event, context, cfnresponse.SUCCESS, response_data) + send_cfn_response( + cfnresponse, event, context, cfnresponse.SUCCESS, response_data + ) except Exception as e: LOGGER.error(f"Error deleting policy: {str(e)}") - _send_cfn_response(event, context, cfnresponse.FAILED, {"Message": str(e)}) + send_cfn_response( + cfnresponse, event, context, cfnresponse.FAILED, {"Message": str(e)} + ) def handle_create_update(event, context): @@ -1263,10 +1255,12 @@ def handle_create_update(event, context): ) if target_changed: _cleanup_previous_target_policies(iam_client, previous_props) - _send_cfn_response(event, context, cfnresponse.SUCCESS, {}) + send_cfn_response(cfnresponse, event, context, cfnresponse.SUCCESS, {}) except Exception as e: LOGGER.error(f"Error creating/attaching policy: {str(e)}") - _send_cfn_response(event, context, cfnresponse.FAILED, {"Message": str(e)}) + send_cfn_response( + cfnresponse, event, context, cfnresponse.FAILED, {"Message": str(e)} + ) def handler(event, context): diff --git a/aws_quickstart/attach_integration_permissions_test.py b/aws_quickstart/attach_integration_permissions_test.py index baa2c463..35d6fca0 100644 --- a/aws_quickstart/attach_integration_permissions_test.py +++ b/aws_quickstart/attach_integration_permissions_test.py @@ -70,11 +70,11 @@ def test_release_embeds_tested_source(self): release = release_path.read_text() self.assertIn( - 'cp datadog_agentless_api_call.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/"', + 'cp datadog_agentless_api_call.py cfn_common.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/"', release, ) self.assertIn( - "embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE", + "embed_python_source_with_common datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE", release, ) diff --git a/aws_quickstart/cfn_common.py b/aws_quickstart/cfn_common.py new file mode 100644 index 00000000..0fb597a1 --- /dev/null +++ b/aws_quickstart/cfn_common.py @@ -0,0 +1,14 @@ +def physical_resource_id(event): + return event.get("PhysicalResourceId") or ( + f"{event['StackId']}/{event['LogicalResourceId']}" + ) + + +def send_cfn_response(cfn_response, event, context, status, response_data): + cfn_response.send( + event, + context, + status, + responseData=response_data, + physicalResourceId=physical_resource_id(event), + ) diff --git a/aws_quickstart/cfn_common_test.py b/aws_quickstart/cfn_common_test.py new file mode 100644 index 00000000..c70afb08 --- /dev/null +++ b/aws_quickstart/cfn_common_test.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 + +from pathlib import Path +import unittest +from unittest.mock import MagicMock + +from cfn_common import physical_resource_id, send_cfn_response + + +def event(**overrides): + value = { + "StackId": "arn:aws:cloudformation:us-east-1:123456789012:stack/test/id", + "LogicalResourceId": "CustomResource", + } + value.update(overrides) + return value + + +class TestPhysicalResourceId(unittest.TestCase): + def test_preserves_existing_id(self): + self.assertEqual( + physical_resource_id(event(PhysicalResourceId="existing-id")), + "existing-id", + ) + + def test_builds_deterministic_id(self): + value = event() + self.assertEqual( + physical_resource_id(value), + f"{value['StackId']}/{value['LogicalResourceId']}", + ) + + +class TestSendCfnResponse(unittest.TestCase): + def test_sends_response_with_physical_resource_id(self): + cfn_response = MagicMock() + value = event() + + send_cfn_response( + cfn_response, + value, + "context", + "SUCCESS", + {"Result": "ok"}, + ) + + cfn_response.send.assert_called_once_with( + value, + "context", + "SUCCESS", + responseData={"Result": "ok"}, + physicalResourceId=f"{value['StackId']}/{value['LogicalResourceId']}", + ) + + +class TestInlineComposition(unittest.TestCase): + def test_shared_helper_composes_with_each_handler(self): + directory = Path(__file__).parent + common = (directory / "cfn_common.py").read_text() + + for filename in ( + "attach_integration_permissions.py", + "accept_operator_subscription.py", + ): + handler = (directory / filename).read_text().replace( + "from cfn_common import send_cfn_response\n", "" + ) + source = f"{common}\n{handler}" + + with self.subTest(filename=filename): + self.assertNotIn("from cfn_common import", source) + compile(source, filename, "exec") + + +if __name__ == "__main__": + unittest.main() diff --git a/aws_quickstart/release.sh b/aws_quickstart/release.sh index 5a242f68..035358e0 100755 --- a/aws_quickstart/release.sh +++ b/aws_quickstart/release.sh @@ -57,6 +57,23 @@ embed_python_source() { ' "$template" < "$source" } +embed_python_source_with_common() { + local template="$1" + local source="$2" + local placeholder="$3" + local composed_source + composed_source=$(mktemp) + + # CloudFormation ZipFile Lambdas are single-file modules, so inline the shared + # helper and remove the local-development import from the handler source. + { + sed -e '$a\' cfn_common.py + sed '/^from cfn_common import send_cfn_response$/d' "$source" + } > "${composed_source}" + embed_python_source "$template" "${composed_source}" "$placeholder" + rm -f "${composed_source}" +} + # Parse flags and optional bucket argument GOV=false PRIVATE_TEMPLATE=false @@ -138,7 +155,7 @@ trap "rm -rf ${TEMP_DIR}" EXIT # Copy all YAML files to temp directory cp *.yaml "${TEMP_DIR}/" -cp datadog_agentless_api_call.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/" +cp datadog_agentless_api_call.py cfn_common.py attach_integration_permissions.py accept_operator_subscription.py "${TEMP_DIR}/" # Change to temp directory for processing cd "${TEMP_DIR}" @@ -165,8 +182,8 @@ for template in main_workflow.yaml main_extended_workflow.yaml main_v2.yaml main fi done -embed_python_source datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE -embed_python_source datadog_integration_permissions.yaml accept_operator_subscription.py ACCEPT_OPERATOR_SUBSCRIPTION_SOURCE +embed_python_source_with_common datadog_integration_permissions.yaml attach_integration_permissions.py ATTACH_INTEGRATION_PERMISSIONS_SOURCE +embed_python_source_with_common datadog_integration_permissions.yaml accept_operator_subscription.py ACCEPT_OPERATOR_SUBSCRIPTION_SOURCE # Process Agentless Scanning templates for template in datadog_agentless_delegate_role.yaml datadog_agentless_scanning.yaml datadog_agentless_delegate_role_snapshot.yaml datadog_integration_autoscaling_policy.yaml datadog_integration_sds_policy.yaml datadog_agentless_delegate_role_stackset.yaml datadog_agentless_saas.yaml; do From 7873b177b15b93f39a584f5e9d6fd6937154a9cc Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Mon, 24 Aug 2026 12:08:54 -0400 Subject: [PATCH 6/6] TON-829: Refine subscription validation --- .../accept_operator_subscription.py | 43 ++------- .../accept_operator_subscription_test.py | 87 +++++++++++++++---- .../datadog_integration_permissions.yaml | 1 - 3 files changed, 78 insertions(+), 53 deletions(-) diff --git a/aws_quickstart/accept_operator_subscription.py b/aws_quickstart/accept_operator_subscription.py index 1ce629b8..1dca8c77 100644 --- a/aws_quickstart/accept_operator_subscription.py +++ b/aws_quickstart/accept_operator_subscription.py @@ -20,7 +20,6 @@ MARKETPLACE_REGION = "us-east-1" ENTITLEMENT_ATTEMPTS = 24 ENTITLEMENT_DELAY_SECONDS = 5 -AWS_READ_TIMEOUT_SECONDS = 20 SDK_CONNECT_TIMEOUT_SECONDS = 5 SDK_READ_TIMEOUT_SECONDS = 10 SDK_REQUEST_BUDGET_SECONDS = 32 @@ -295,8 +294,7 @@ def requested_terms(discovery_client, offer_id, *, deadline=None): offerId=offer_id, ) ) - term_ids = [] - seen = set() + term_ids_by_name = {} supported = {"legalTerm", "supportTerm"} for term in terms: present = [name for name in supported if name in term] @@ -308,7 +306,7 @@ def requested_terms(discovery_client, offer_id, *, deadline=None): offer_id=offer_id, ) term_name = present[0] - if term_name in seen: + if term_name in term_ids_by_name: raise SubscriptionError( "offer_terms", "invalid_terms", @@ -323,10 +321,9 @@ def requested_terms(discovery_client, offer_id, *, deadline=None): "The Datadog Operator Marketplace offer contains a term without an identifier", offer_id=offer_id, ) - seen.add(term_name) - term_ids.append(term_id) + term_ids_by_name[term_name] = term_id - if seen != supported: + if term_ids_by_name.keys() != supported: raise SubscriptionError( "offer_terms", "invalid_terms", @@ -334,7 +331,7 @@ def requested_terms(discovery_client, offer_id, *, deadline=None): "legal and one support term", offer_id=offer_id, ) - return [{"id": term_id} for term_id in sorted(term_ids)] + return [{"id": term_id} for term_id in sorted(term_ids_by_name.values())] def _require_zero(value, field, *, required=False): @@ -461,8 +458,8 @@ def create_and_accept_agreement( marketplace_agreement_request_id=agreement_request_id, ) + _require_sdk_request_budget(deadline, "accept_agreement_request") try: - _require_sdk_request_budget(deadline, "accept_agreement_request") accepted = agreement_client.accept_agreement_request( agreementRequestId=agreement_request_id ) @@ -561,7 +558,7 @@ def wait_for_entitlement( for attempt in range(attempts): if ( deadline is not None - and time.monotonic() + AWS_READ_TIMEOUT_SECONDS >= deadline + and time.monotonic() + SDK_REQUEST_BUDGET_SECONDS >= deadline ): break entitlement = entitlement_status( @@ -598,7 +595,7 @@ def wait_for_entitlement( if attempt + 1 < attempts: if ( deadline is not None - and time.monotonic() + delay + AWS_READ_TIMEOUT_SECONDS >= deadline + and time.monotonic() + delay + SDK_REQUEST_BUDGET_SECONDS >= deadline ): break time.sleep(delay) @@ -667,7 +664,6 @@ def handler(event, context): request_type = event["RequestType"] properties = event["ResourceProperties"] account_id = properties.get("AccountId") - partition = properties.get("Partition", "aws") if request_type == "Delete": _log( "cloudformation_delete", @@ -684,29 +680,6 @@ def handler(event, context): ) return - if partition != "aws": - error = SubscriptionError( - "partition_validation", - "unsupported_partition", - "Datadog Operator Marketplace subscription acceptance is supported " - "only in the commercial AWS partition", - ) - _log( - error.stage, - "failed", - error.reason, - account_id=account_id, - error=str(error), - ) - send_cfn_response( - cfnresponse, - event, - context, - cfnresponse.FAILED, - {"Message": str(error)}, - ) - return - try: remaining_seconds = context.get_remaining_time_in_millis() / 1000 deadline = time.monotonic() + max( diff --git a/aws_quickstart/accept_operator_subscription_test.py b/aws_quickstart/accept_operator_subscription_test.py index d0be9dba..7d135950 100644 --- a/aws_quickstart/accept_operator_subscription_test.py +++ b/aws_quickstart/accept_operator_subscription_test.py @@ -75,7 +75,7 @@ def free_offer(offer_id="offer-1", **overrides): return value -def event(request_type="Create", partition="aws"): +def event(request_type="Create"): return { "RequestType": request_type, "RequestId": "request-1", @@ -83,7 +83,6 @@ def event(request_type="Create", partition="aws"): "LogicalResourceId": "DatadogOperatorSubscriptionFunctionTrigger", "ResourceProperties": { "AccountId": "123456789012", - "Partition": partition, }, } @@ -95,7 +94,16 @@ def test_template_embeds_subscription_source(self): ).read_text() self.assertEqual(template.count(""), 1) - self.assertIn(" IncludeEKS:", template) + self.assertIn( + "Conditions:\n" + " IncludeEKS:\n" + " Fn::And:\n" + " - Fn::Equals:\n" + " - !Ref AWS::Partition\n" + " - aws\n" + " - Fn::Not:\n", + template, + ) self.assertIn( " InstrumentationResourceTypes:\n Type: CommaDelimitedList", template, @@ -112,10 +120,6 @@ def test_template_embeds_subscription_source(self): "!Ref InstrumentationResourceTypes]", template, ) - self.assertIn( - " - Fn::Equals:\n - !Ref AWS::Partition\n - aws", - template, - ) for resource in ( "DatadogOperatorSubscriptionLambdaExecutionRole", "DatadogOperatorSubscriptionFunction", @@ -454,6 +458,36 @@ def test_recovers_ambiguous_acceptance( "agreement-1", ) + @patch("accept_operator_subscription.time.monotonic", side_effect=[0, 100]) + @patch("accept_operator_subscription.find_active_agreement") + @patch("accept_operator_subscription.find_free_offer") + @patch("accept_operator_subscription.requested_terms") + def test_does_not_recover_when_acceptance_was_not_attempted( + self, + mock_terms, + mock_offer, + mock_find_active, + _mock_monotonic, + ): + mock_offer.return_value = free_offer() + mock_terms.return_value = [{"id": "legal"}, {"id": "support"}] + agreement = MagicMock() + agreement.create_agreement_request.return_value = { + "agreementRequestId": "request-1", + "chargeSummary": {"newAgreementValue": "0"}, + } + + with self.assertRaisesRegex(RuntimeError, "accept_agreement_request"): + create_and_accept_agreement( + event(), + MagicMock(), + agreement, + deadline=120, + ) + + agreement.accept_agreement_request.assert_not_called() + mock_find_active.assert_not_called() + class TestEntitlements(unittest.TestCase): def test_returns_only_operator_entitlement_across_pages(self): @@ -506,7 +540,7 @@ def test_fails_terminal_status_and_timeout(self): with self.assertRaisesRegex(SubscriptionError, "Timed out"): wait_for_entitlement(MagicMock(), "agreement-1", attempts=1) - @patch("accept_operator_subscription.time.monotonic", return_value=90) + @patch("accept_operator_subscription.time.monotonic", return_value=70) def test_stops_before_deadline_without_starting_request(self, _mock_monotonic): client = MagicMock() @@ -515,6 +549,34 @@ def test_stops_before_deadline_without_starting_request(self, _mock_monotonic): client.get_paginator.assert_not_called() + @patch("accept_operator_subscription.time.sleep") + @patch("accept_operator_subscription.entitlement_status") + @patch("accept_operator_subscription.time.monotonic", side_effect=[0, 65]) + def test_preserves_last_status_when_next_request_exceeds_budget( + self, + _mock_monotonic, + mock_status, + mock_sleep, + ): + mock_status.return_value = { + "status": "PENDING", + "statusReasonCode": "PROVISIONING_IN_PROGRESS", + } + + with self.assertRaises(SubscriptionError) as raised: + wait_for_entitlement( + MagicMock(), + "agreement-1", + attempts=2, + delay=5, + deadline=100, + ) + + self.assertEqual(raised.exception.reason, "entitlement_timeout") + self.assertIn("status=PENDING", str(raised.exception)) + mock_status.assert_called_once() + mock_sleep.assert_not_called() + class TestAPIFailureStages(unittest.TestCase): def assert_stage(self, expected_stage, operation): @@ -636,15 +698,6 @@ def test_delete_retains_agreement_without_aws_calls(self, mock_session): self.response().kwargs["responseData"], {"AgreementRetained": True} ) - def test_fails_eks_on_unsupported_partition(self): - handler(event(partition="aws-us-gov"), self.context) - - self.assertEqual(self.response().args[2], "FAILED") - self.assertIn( - "commercial AWS partition", - self.response().kwargs["responseData"]["Message"], - ) - @patch("accept_operator_subscription.time.monotonic", return_value=100) @patch("accept_operator_subscription.ensure_subscription") def test_returns_agreement_details(self, mock_ensure, _mock_monotonic): diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 51585169..45792f71 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -253,4 +253,3 @@ Resources: ServiceToken: !GetAtt DatadogOperatorSubscriptionFunction.Arn SubscriptionSchemaVersion: "1" AccountId: !Ref AWS::AccountId - Partition: !Sub "${AWS::Partition}"