feat: Included event source mappings for lambda service

Author: Marco Franceschi

src/services/lambda/data.ts

@@ -9,7 +9,9 @@ import Lambda, {
   GetFunctionConcurrencyRequest,
   GetFunctionConcurrencyResponse,
   ReservedConcurrentExecutions,
-  GetPolicyResponse
+  GetPolicyResponse,
+  ListEventSourceMappingsResponse,
+  EventSourceMappingConfiguration
 } from 'aws-sdk/clients/lambda'
 import { AWSError } from 'aws-sdk/lib/error'
 import { Config } from 'aws-sdk/lib/config'
@@ -35,6 +37,7 @@ export interface RawAwsLambdaFunction extends FunctionConfiguration {
     Policy?: string
     RevisionId?: string
   }
+  EventSourceMappings?: EventSourceMappingConfiguration[]
 }
 
 const listFunctionsForRegion = async ({
@@ -137,7 +140,7 @@ const getResourceTags = async (lambda: Lambda, arn: string): Promise<TagMap> =>
     }
   })
 
-  const getLambdaPolicy = async (lambda: Lambda, arn: string): Promise<{ Policy?: string; RevisionId?: string }> =>
+const getLambdaPolicy = async (lambda: Lambda, arn: string): Promise<{ Policy?: string; RevisionId?: string }> =>
   new Promise(resolve => {
     try {
       lambda.getPolicy(
@@ -159,6 +162,29 @@ const getResourceTags = async (lambda: Lambda, arn: string): Promise<TagMap> =>
     }
   })
 
+const getEventSourceMappings = async (lambda: Lambda, arn: string): Promise<EventSourceMappingConfiguration[]> =>
+  new Promise(resolve => {
+    try {
+      lambda.listEventSourceMappings(
+        { FunctionName: arn },
+        (err: AWSError, data: ListEventSourceMappingsResponse) => {
+          if (err) {
+            errorLog.generateAwsErrorLog({
+              functionName: 'lambda:listEventSourceMappings',
+              err,
+            })
+            resolve([])
+          }
+          const { EventSourceMappings = [] } = data || {}
+
+          resolve(EventSourceMappings)
+        }
+      )
+    } catch (error) {
+      resolve([])
+    }
+  })
+
 export default async ({
   regions,
   config,
@@ -201,14 +227,16 @@ export default async ({
     // get all tags and policy for each Lambda
     lambdaData.map(({ FunctionArn: arn, region }, idx) => {
       const lambda = new Lambda({ ...config, region, endpoint })
-      const tagsAndPolicyPromise = new Promise<void>(async resolveData => {
+      const additionalMetadataPromise = new Promise<void>(async resolveData => {
         const envTags: TagMap = await getResourceTags(lambda, arn)
         lambdaData[idx].Tags = envTags
         const policy = await getLambdaPolicy(lambda, arn)
         lambdaData[idx].PolicyData = policy
+        const eventSourceMappings = await getEventSourceMappings(lambda, arn)
+        lambdaData[idx].EventSourceMappings = eventSourceMappings
         resolveData()
       })
-      tagsPromises.push(tagsAndPolicyPromise)
+      tagsPromises.push(additionalMetadataPromise)
     })
 
     logger.debug(lt.gettingLambdaTags)

src/services/lambda/format.ts

@@ -1,8 +1,10 @@
+import { generateUniqueId } from '@cloudgraph/sdk'
 import isEmpty from 'lodash/isEmpty'
 import t from '../../properties/translations'
-import { AwsLambda } from '../../types/generated'
+import { AwsLambda, AwsLambdaEventSourceMappings } from '../../types/generated'
 import { formatTagsFromMap, formatIamJsonPolicy } from '../../utils/format'
 import { RawAwsLambdaFunction } from './data'
+import { EventSourceMappingConfiguration } from 'aws-sdk/clients/lambda'
 
 /**
  * Lambda
@@ -33,6 +35,7 @@ export default ({
     reservedConcurrentExecutions: rawReservedConcurrentExecutions,
     VpcConfig: vpcConfig,
     PolicyData: { Policy: policy = '', RevisionId: policyRevisionId = '' },
+    EventSourceMappings: eventSourceMappings = []
   } = rawData
   const environmentVariables = []
   const secretNames = [t.pass, t.secret, t.private, t.cert]
@@ -68,6 +71,62 @@ export default ({
     securityGroupIds: vpcConfig?.SecurityGroupIds,
   }
 
+  const formatEventSourceMappings = (
+    eventSourceMappings?: EventSourceMappingConfiguration[]
+  ): AwsLambdaEventSourceMappings[] => {
+    return (
+      eventSourceMappings?.map(e => ({
+        id: generateUniqueId({
+          arn,
+          ...e,
+        }),
+        uuid: e.UUID,
+        startingPosition: e.StartingPosition,
+        batchSize: e.BatchSize,
+        maximumBatchingWindowInSeconds: e.MaximumBatchingWindowInSeconds,
+        parallelizationFactor: e.ParallelizationFactor,
+        eventSourceArn: e.EventSourceArn,
+        filterCriteria: e.FilterCriteria?.Filters?.map(f => f.Pattern) || [],
+        functionArn: e.FunctionArn,
+        lastModified: e.LastModified?.toISOString(),
+        lastProcessingResult: e.LastProcessingResult,
+        state: e.State,
+        stateTransitionReason: e.StateTransitionReason,
+        destinationConfig: {
+          id: generateUniqueId({
+            arn,
+            ...e.DestinationConfig,
+
+          }),
+          OnSuccess: e.DestinationConfig?.OnSuccess?.Destination,
+          OnFailure: e.DestinationConfig?.OnFailure?.Destination
+
+        },
+        topics: e.Topics,
+        queues: e.Queues,
+        maximumRecordAgeInSeconds: e.MaximumRecordAgeInSeconds,
+        bisectBatchOnFunctionError: e.BisectBatchOnFunctionError,
+        maximumRetryAttempts: e.MaximumRetryAttempts,
+        tumblingWindowInSeconds: e.TumblingWindowInSeconds,
+        functionResponseTypes: e.FunctionResponseTypes,
+        amazonManagedKafkaEventSourceConfig: {
+          id: generateUniqueId({
+            arn,
+            ...e.AmazonManagedKafkaEventSourceConfig
+          }),
+          consumerGroupId: e.AmazonManagedKafkaEventSourceConfig?.ConsumerGroupId
+        },
+        selfManagedKafkaEventSourceConfig: {
+          id: generateUniqueId({
+            arn,
+            ...e.SelfManagedKafkaEventSourceConfig
+          }),
+          consumerGroupId: e.SelfManagedKafkaEventSourceConfig?.ConsumerGroupId
+        }
+      })) || []
+    )
+  }
+
   return {
     accountId: account,
     arn,
@@ -90,5 +149,6 @@ export default ({
     rawPolicy: policy,
     policy: formatIamJsonPolicy(policy),
     tags: formatTagsFromMap(Tags),
+    eventSourceMappings: formatEventSourceMappings(eventSourceMappings)
   }
 }

src/services/lambda/schema.graphql

@@ -1,3 +1,67 @@
+type awsLambdaEventSourceConfig
+  @generate(
+    query: { get: false, query: true, aggregate: false }
+    mutation: { add: false, delete: false }
+    subscription: false
+  ) {
+  id: String! @id
+  consumerGroupId: String @search(by: [hash, regexp])
+}
+
+type awsLambdaSourceAccessConfiguration
+  @generate(
+    query: { get: false, query: true, aggregate: false }
+    mutation: { add: false, delete: false }
+    subscription: false
+  ) {
+  id: String! @id
+  type: String @search(by: [hash, regexp])
+  uri: String @search(by: [hash, regexp])
+}
+
+type awsLambdaDestinationConfig
+  @generate(
+    query: { get: false, query: true, aggregate: false }
+    mutation: { add: false, delete: false }
+    subscription: false
+  ) {
+  id: String! @id
+  OnSuccess: String @search(by: [hash, regexp])
+  OnFailure: String @search(by: [hash, regexp])
+}
+
+type awsLambdaEventSourceMappings
+  @generate(
+    query: { get: false, query: true, aggregate: false }
+    mutation: { add: false, delete: false }
+    subscription: false
+  ) {
+  id: String! @id
+  uuid: String @search(by: [hash, regexp])
+  startingPosition: String @search(by: [hash, regexp])
+  batchSize: Int
+  maximumBatchingWindowInSeconds: Int
+  parallelizationFactor: Int
+  eventSourceArn: String @search(by: [hash, regexp])
+  filterCriteria: [String] @search(by: [hash, regexp])
+  functionArn: String @search(by: [hash, regexp])
+  lastModified: DateTime @search(by: [day])
+  lastProcessingResult: String @search(by: [hash, regexp])
+  state: String @search(by: [hash, regexp])
+  stateTransitionReason: String @search(by: [hash, regexp])
+  destinationConfig: awsLambdaDestinationConfig
+  topics: [String] @search(by: [hash, regexp])
+  queues: [String] @search(by: [hash, regexp])
+  maximumRecordAgeInSeconds: Int
+  bisectBatchOnFunctionError: Boolean
+  maximumRetryAttempts: Int
+  tumblingWindowInSeconds: Int
+  functionResponseTypes: [String] @search(by: [hash, regexp])
+  amazonManagedKafkaEventSourceConfig: awsLambdaEventSourceConfig
+  selfManagedKafkaEventSourceConfig: awsLambdaEventSourceConfig
+}
+
+
 type awsLambda implements awsBaseService @key(fields: "arn") {
   description: String @search(by: [hash, regexp, fulltext])
   handler: String @search(by: [hash, regexp])
@@ -15,6 +79,7 @@ type awsLambda implements awsBaseService @key(fields: "arn") {
   policyRevisionId: String @search(by: [hash, regexp])
   rawPolicy: String @search(by: [hash, regexp])
   policy: awsIamJSONPolicy
+  eventSourceMappings: [awsLambdaEventSourceMappings]
   tags: [awsRawTag]
   kms: [awsKms] @hasInverse(field: lambda)
   securityGroups: [awsSecurityGroup] @hasInverse(field: lambda)

src/types/generated.ts

@@ -3410,6 +3410,7 @@ export type AwsLambda = AwsBaseService & {
   cognitoUserPools?: Maybe<Array<Maybe<AwsCognitoUserPool>>>;
   description?: Maybe<Scalars['String']>;
   environmentVariables?: Maybe<Array<Maybe<AwsLambdaEnvironmentVariable>>>;
+  eventSourceMappings?: Maybe<Array<Maybe<AwsLambdaEventSourceMappings>>>;
   handler?: Maybe<Scalars['String']>;
   iamRole?: Maybe<Array<Maybe<AwsIamRole>>>;
   kms?: Maybe<Array<Maybe<AwsKms>>>;
@@ -3434,12 +3435,57 @@ export type AwsLambda = AwsBaseService & {
   vpcConfig?: Maybe<AwsLambdaVpcConfig>;
 };
 
+export type AwsLambdaDestinationConfig = {
+  OnFailure?: Maybe<Scalars['String']>;
+  OnSuccess?: Maybe<Scalars['String']>;
+  id: Scalars['String'];
+};
+
 export type AwsLambdaEnvironmentVariable = {
   id: Scalars['String'];
   key: Scalars['String'];
   value?: Maybe<Scalars['String']>;
 };
 
+export type AwsLambdaEventSourceConfig = {
+  consumerGroupId?: Maybe<Scalars['String']>;
+  id: Scalars['String'];
+};
+
+export type AwsLambdaEventSourceMappings = {
+  amazonManagedKafkaEventSourceConfig?: Maybe<AwsLambdaEventSourceConfig>;
+  batchSize?: Maybe<Scalars['Int']>;
+  bisectBatchOnFunctionError?: Maybe<Scalars['Boolean']>;
+  destinationConfig?: Maybe<AwsLambdaDestinationConfig>;
+  eventSourceArn?: Maybe<Scalars['String']>;
+  filterCriteria?: Maybe<Array<Maybe<Scalars['String']>>>;
+  functionArn?: Maybe<Scalars['String']>;
+  functionResponseTypes?: Maybe<Array<Maybe<Scalars['String']>>>;
+  id: Scalars['String'];
+  lastModified?: Maybe<Scalars['DateTime']>;
+  lastProcessingResult?: Maybe<Scalars['String']>;
+  maximumBatchingWindowInSeconds?: Maybe<Scalars['Int']>;
+  maximumRecordAgeInSeconds?: Maybe<Scalars['Int']>;
+  maximumRetryAttempts?: Maybe<Scalars['Int']>;
+  parallelizationFactor?: Maybe<Scalars['Int']>;
+  queues?: Maybe<Array<Maybe<Scalars['String']>>>;
+  selfManagedEventSource?: Maybe<Array<Maybe<Scalars['String']>>>;
+  selfManagedKafkaEventSourceConfig?: Maybe<AwsLambdaEventSourceConfig>;
+  sourceAccessConfigurations?: Maybe<Array<Maybe<AwsLambdaSourceAccessConfiguration>>>;
+  startingPosition?: Maybe<Scalars['String']>;
+  state?: Maybe<Scalars['String']>;
+  stateTransitionReason?: Maybe<Scalars['String']>;
+  topics?: Maybe<Array<Maybe<Scalars['String']>>>;
+  tumblingWindowInSeconds?: Maybe<Scalars['Int']>;
+  uuid?: Maybe<Scalars['String']>;
+};
+
+export type AwsLambdaSourceAccessConfiguration = {
+  id: Scalars['String'];
+  type?: Maybe<Scalars['String']>;
+  uri?: Maybe<Scalars['String']>;
+};
+
 export type AwsLambdaVpcConfig = {
   securityGroupIds?: Maybe<Array<Maybe<Scalars['String']>>>;
   subnetIds?: Maybe<Array<Maybe<Scalars['String']>>>;