feat: make Logging property for GraphQLApi (#2971)

Author: pradrx

samtranslator/internal/model/appsync.py

@@ -25,19 +25,27 @@ class DynamoDBConfigType(TypedDict, total=False):
     DeltaSyncConfig: DeltaSyncConfigType
 
 
+class LogConfigType(TypedDict, total=False):
+    CloudWatchLogsRoleArn: Intrinsicable[str]
+    ExcludeVerboseContent: bool
+    FieldLogLevel: str
+
+
 class GraphQLApi(Resource):
     resource_type = "AWS::AppSync::GraphQLApi"
     property_types = {
         "Name": GeneratedProperty(),
         "Tags": GeneratedProperty(),
         "XrayEnabled": GeneratedProperty(),
         "AuthenticationType": GeneratedProperty(),
+        "LogConfig": GeneratedProperty(),
     }
 
     Name: str
     AuthenticationType: str
     Tags: Optional[List[Dict[str, Any]]]
     XrayEnabled: Optional[bool]
+    LogConfig: Optional[LogConfigType]
 
     runtime_attrs = {"api_id": lambda self: fnGetAtt(self.logical_id, "ApiId")}
 

samtranslator/internal/schema_source/aws_serverless_graphqlapi.py

@@ -1,8 +1,8 @@
-from typing import Optional
+from typing import Optional, Union
 
 from typing_extensions import Literal
 
-from samtranslator.internal.schema_source.common import BaseModel, DictStrAny, get_prop
+from samtranslator.internal.schema_source.common import BaseModel, DictStrAny, PassThroughProp, get_prop
 
 properties = get_prop("sam-resource-graphqlapi")
 
@@ -12,13 +12,20 @@ class Auth(BaseModel):
     Type: str
 
 
+class Logging(BaseModel):
+    CloudWatchLogsRoleArn: Optional[str]
+    ExcludeVerboseContent: Optional[PassThroughProp]
+    FieldLogLevel: Optional[str]
+
+
 class Properties(BaseModel):
     Auth: Auth
     Tags: Optional[DictStrAny]
     Name: Optional[str]
     XrayEnabled: Optional[bool]
     SchemaInline: Optional[str]
     SchemaUri: Optional[str]
+    Logging: Optional[Union[Logging, bool]]
 
 
 class Resource(BaseModel):

samtranslator/model/sam_resources.py

@@ -16,6 +16,7 @@
     DynamoDBConfigType,
     GraphQLApi,
     GraphQLSchema,
+    LogConfigType,
 )
 from samtranslator.internal.types import GetManagedPolicyMap
 from samtranslator.intrinsics.resolver import IntrinsicsResolver
@@ -2136,6 +2137,7 @@ class SamGraphQLApi(SamResourceMacro):
         "Auth": Property(True, IS_DICT),
         "SchemaInline": Property(False, IS_STR),
         "SchemaUri": Property(False, IS_STR),
+        "Logging": Property(False, one_of(IS_DICT, is_type(bool))),
     }
 
     Auth: Dict[str, Any]
@@ -2144,16 +2146,21 @@ class SamGraphQLApi(SamResourceMacro):
     Name: Optional[str]
     SchemaInline: Optional[str]
     SchemaUri: Optional[str]
+    Logging: Optional[Union[Dict[str, Any], bool]]
 
     @cw_timer
     def to_cloudformation(self, **kwargs: Any) -> List[Resource]:
-        appsync_api = self._construct_appsync_api()
+        appsync_api, cloudwatch_role = self._construct_appsync_api_resources()
         appsync_schema = self._construct_appsync_schema(appsync_api.get_runtime_attr("api_id"))
+
         resources: List[Resource] = [appsync_api, appsync_schema]
 
+        if cloudwatch_role:
+            resources.append(cloudwatch_role)
+
         return resources
 
-    def _construct_appsync_api(self) -> GraphQLApi:
+    def _construct_appsync_api_resources(self) -> Tuple[GraphQLApi, Optional[IAMRole]]:
         api = GraphQLApi(logical_id=self.logical_id, depends_on=self.depends_on, attributes=self.resource_attributes)
 
         api.AuthenticationType = sam_expect(self.Auth.get("Type"), self.logical_id, "Auth.Type").to_be_a_string()
@@ -2163,7 +2170,63 @@ def _construct_appsync_api(self) -> GraphQLApi:
         if self.Tags:
             api.Tags = get_tag_list(self.Tags)
 
-        return api
+        # Logging has 3 possible types: dict, bool, and None.
+        # GraphQLApi will not include logging if and only if the user explicity sets Logging as false boolean.
+        # It will for every other value (including true boolean which is essentially same as None).
+        if isinstance(self.Logging, bool) and self.Logging is False:
+            return api, None
+
+        api.LogConfig, cloudwatch_role = self._parse_logging_properties()
+
+        return api, cloudwatch_role
+
+    def _create_logging_default(self) -> Tuple[LogConfigType, IAMRole]:
+        """
+        Create a default logging configuration.
+
+        This function is used when "Logging" property is a False boolean or NoneType.
+        """
+        log_config: LogConfigType = {}
+        log_config["FieldLogLevel"] = "ALL"
+        cloudwatch_role = self._construct_cloudwatch_role()
+        log_config["CloudWatchLogsRoleArn"] = cloudwatch_role.get_runtime_attr("arn")
+
+        return log_config, cloudwatch_role
+
+    def _parse_logging_properties(self) -> Tuple[LogConfigType, Optional[IAMRole]]:
+        """Parse logging properties from SAM template, and use defaults if required keys dont exist."""
+        if not isinstance(self.Logging, dict):
+            return self._create_logging_default()
+
+        log_config: LogConfigType = {}
+
+        if "ExcludeVerboseContent" in self.Logging:
+            log_config["ExcludeVerboseContent"] = self.Logging["ExcludeVerboseContent"]
+
+        log_config["FieldLogLevel"] = self.Logging.get("FieldLogLevel", "ALL")
+        log_config["CloudWatchLogsRoleArn"] = self.Logging.get("CloudWatchLogsRoleArn", None)
+
+        if log_config["CloudWatchLogsRoleArn"]:
+            return log_config, None
+
+        cloudwatch_role = self._construct_cloudwatch_role()
+        log_config["CloudWatchLogsRoleArn"] = cloudwatch_role.get_runtime_attr("arn")
+
+        return log_config, cloudwatch_role
+
+    def _construct_cloudwatch_role(self) -> IAMRole:
+        role = IAMRole(
+            logical_id=f"{self.logical_id}CloudWatchRole",
+            depends_on=self.depends_on,
+            attributes=self.resource_attributes,
+        )
+        role.AssumeRolePolicyDocument = IAMRolePolicies.construct_assume_role_policy_for_service_principal(
+            "appsync.amazonaws.com"
+        )
+        role.ManagedPolicyArns = [
+            {"Fn::Sub": "arn:${AWS::Partition}:iam::aws:policy/service-role/AWSAppSyncPushToCloudWatchLogs"}
+        ]
+        return role
 
     def _construct_appsync_schema(self, api_id: Intrinsicable[str]) -> GraphQLSchema:
         schema = GraphQLSchema(

samtranslator/schema/schema.json

@@ -195464,6 +195464,24 @@
       "title": "Location",
       "type": "object"
     },
+    "Logging": {
+      "additionalProperties": false,
+      "properties": {
+        "CloudWatchLogsRoleArn": {
+          "title": "Cloudwatchlogsrolearn",
+          "type": "string"
+        },
+        "ExcludeVerboseContent": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
+        "FieldLogLevel": {
+          "title": "Fieldloglevel",
+          "type": "string"
+        }
+      },
+      "title": "Logging",
+      "type": "object"
+    },
     "MQEvent": {
       "additionalProperties": false,
       "properties": {
@@ -199349,6 +199367,17 @@
         "Auth": {
           "$ref": "#/definitions/samtranslator__internal__schema_source__aws_serverless_graphqlapi__Auth"
         },
+        "Logging": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/Logging"
+            },
+            {
+              "type": "boolean"
+            }
+          ],
+          "title": "Logging"
+        },
         "Name": {
           "title": "Name",
           "type": "string"

schema_source/sam.schema.json

@@ -1958,6 +1958,24 @@
       "title": "Location",
       "type": "object"
     },
+    "Logging": {
+      "additionalProperties": false,
+      "properties": {
+        "CloudWatchLogsRoleArn": {
+          "title": "Cloudwatchlogsrolearn",
+          "type": "string"
+        },
+        "ExcludeVerboseContent": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
+        "FieldLogLevel": {
+          "title": "Fieldloglevel",
+          "type": "string"
+        }
+      },
+      "title": "Logging",
+      "type": "object"
+    },
     "MQEvent": {
       "additionalProperties": false,
       "properties": {
@@ -5748,6 +5766,17 @@
         "Auth": {
           "$ref": "#/definitions/samtranslator__internal__schema_source__aws_serverless_graphqlapi__Auth"
         },
+        "Logging": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/Logging"
+            },
+            {
+              "type": "boolean"
+            }
+          ],
+          "title": "Logging"
+        },
         "Name": {
           "title": "Name",
           "type": "string"

tests/translator/input/error_graphqlapi.yaml

@@ -0,0 +1,34 @@
+Transform: AWS::Serverless-2016-10-31
+Resources:
+  NoAuthAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      XrayEnabled: true
+      Tags:
+        Key1: Value1
+        Key2: Value2
+
+  NoSchemaPropertiesAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      XrayEnabled: true
+      Auth:
+        Type: AWS_IAM
+      Tags:
+        key1: value1
+        key2: value2
+
+  BothSchemaInlineAndUriAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      SchemaInline: |
+        type Mutation {
+          addTodo(id: ID!, name: String, description: String, priority: Int): Todo
+        }
+      SchemaUri: https://bucket-name.s3.region-code.amazonaws.com/key-name
+      XrayEnabled: true
+      Auth:
+        Type: AWS_IAM
+      Tags:
+        key1: value1
+        key2: value2

tests/translator/input/error_graphqlapi_no_auth.yaml

@@ -7,7 +7,6 @@ Resources:
         type Mutation {
           addTodo(id: ID!, name: String, description: String, priority: Int): Todo
         }
-      SchemaUri: https://bucket-name.s3.region-code.amazonaws.com/key-name
       XrayEnabled: true
       Auth:
         Type: AWS_IAM

tests/translator/input/error_graphqlapi_both_schema_inline_and_uri.yaml renamed to tests/translator/input/graphqlapi_default_logging.yaml

@@ -0,0 +1,19 @@
+Transform: AWS::Serverless-2016-10-31
+Resources:
+  SuperCoolAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      SchemaInline: |
+        type Mutation {
+          addTodo(id: ID!, name: String, description: String, priority: Int): Todo
+        }
+      XrayEnabled: true
+      Auth:
+        Type: AWS_IAM
+      Tags:
+        key1: value1
+        key2: value2
+      Logging:
+        CloudWatchLogsRoleArn: some-arn
+        FieldLogLevel: ERROR
+        ExcludeVerboseContent: true

tests/translator/input/graphqlapi_logging_defined.yaml

@@ -0,0 +1,17 @@
+# This is not an expected user use case (they should just not define Logging if they want defaults), but still testing.
+Transform: AWS::Serverless-2016-10-31
+Resources:
+  SuperCoolAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      SchemaInline: |
+        type Mutation {
+          addTodo(id: ID!, name: String, description: String, priority: Int): Todo
+        }
+      XrayEnabled: true
+      Auth:
+        Type: AWS_IAM
+      Tags:
+        key1: value1
+        key2: value2
+      Logging: true

tests/translator/input/graphqlapi_logging_true.yaml

@@ -3,9 +3,14 @@ Resources:
   SuperCoolAPI:
     Type: AWS::Serverless::GraphQLApi
     Properties:
+      SchemaInline: |
+        type Mutation {
+          addTodo(id: ID!, name: String, description: String, priority: Int): Todo
+        }
       XrayEnabled: true
       Auth:
         Type: AWS_IAM
       Tags:
         key1: value1
         key2: value2
+      Logging: false