
subaud is authored by Court Schuett (he/him). While Court is an 8090 Inc employee, opinions and statements on this blog are his own.


Court Schuett


This demo will guide you through the process of deploying a [gRPC](https://grpc.io/) server that is capable of receiving streaming audio from a gRPC client and processing that audio with [Amazon Transcribe](https://aws.amazon.com/transcribe/) and returning the result to the client.

[GitHub Repo](https://github.com/schuettc/node-grpc-demo)

## Server Infrastructure

![gRPCServerOverview](/images/blog/node-grpc-server/gRPCServerOverview.png)

This demo will deploy a [Fargate](https://aws.amazon.com/fargate/) container that runs the gRPC server. This gRPC server will be available through an [Application Load Balancer](https://aws.amazon.com/elasticloadbalancing/application-load-balancer/) that is enabled with HTTPS through a [Certificate](https://docs.aws.amazon.com/acm/latest/userguide/gs-acm-request-public.html) generated through [AWS Certificate Manager](https://aws.amazon.com/certificate-manager/).

### Certificate Manager

```typescript
this.hostedZone = HostedZone.fromLookup(this, 'HostedZone', {
  domainName: props.domainName,
});

this.certificate = new Certificate(this, 'Certificate', {
  domainName: `grpc.${props.domainName}`,
  validation: CertificateValidation.fromDns(this.hostedZone),
});
```

Using a provided domain name, the CDK will lookup the associated [HostedZone](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/hosted-zones-working-with.html) and create a Certificate that will be associated with the Application Load Balancer.

### Application Load Balancer

```typescript
const grpcServerTargetGroup = new ApplicationTargetGroup(
  this,
  'gRPCServerTargetGroup',
  {
    vpc: props.vpc,
    port: 50051,
    protocol: ApplicationProtocol.HTTP,
    protocolVersion: ApplicationProtocolVersion.GRPC,
    targets: [this.fargateService],
    healthCheck: {
      healthyGrpcCodes: '12',
    },
  },
);

this.applicationLoadBalancer.addListener('fargateListener', {
  port: 50051,
  protocol: ApplicationProtocol.HTTPS,
  certificates: [ListenerCertificate.fromCertificateManager(props.certificate)],
  open: true,
  defaultTargetGroups: [grpcServerTargetGroup],
});

props.applicationLoadBalancerSecurityGroup.addIngressRule(
  Peer.anyIpv4(),
  Port.tcp(50051),
);
```

This will create an Application Load Balancer [Target Group](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-target-groups.html) that uses `ApplicationProtocolVersion.GRPC` as the protocol version. Additionally, the `healthCheck` will look for gRPC codes in the response from the gRPC server. The [Listener](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-listeners.html) associated with the Application Load Balancer uses HTTPS and the previously created Certificate. The Listener will use port `50051` and forward to port `50051` on the Fargate task. This port is also opened on the Application Load Balancer Security Group.

### Route 53

```typescript
new ARecord(this, 'grpcARecord', {
  zone: props.hostedZone,
  recordName: 'grpc',
  target: RecordTarget.fromAlias(
    new LoadBalancerTarget(this.applicationLoadBalancer),
  ),
});
```

Finally, an [A Record](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/ResourceRecordTypes.html#AFormat) will be created that points to the Application Load Balancer alias use the record name `grpc`. This will create a record `grpc.example.com` that can be used with the Application Load Balancer.

## gRPC Server

### Server initialization

```typescript
function main() {
  const server = new grpc.Server();
  server.addService(audio_proto.AudioStreamer.service, { streamAudio });
  server.bindAsync(
    '0.0.0.0:50051',
    grpc.ServerCredentials.createInsecure(),
    () => {
      server.start();
      console.log('gRPC server started on port 50051');
    },
  );
}

main();
```

The main function of the server code, is used to create the server that is listening on port `50051` with HTTP. The Application Load Balancer will provide the TLS security so HTTP can be used here.

### streamAudio Function

```typescript
function streamAudio(call: grpc.ServerWritableStream<any, any>) {
  console.log('Streaming audio data received from client.');

  const audioStream = new PassThrough();

  startTranscription(audioStream, call).catch((error) => {
    console.error('Transcription error:', error);
    call.write({ message: error });
  });

  call.on('data', (data: any) => {
    audioStream.write(data.audioData);
  });

  call.on('end', async () => {
    console.log('Streaming completed.');
    audioStream.end();
  });

  audioStream.on('end', () => {
    console.log('Transcribing ended.');
  });
}
```

Now, when a stream arrives, the `streamAudio` function will begin processing the stream. In this example, it involves three major steps:

1. Create `audioStream` PassThrough Stream
2. Start `startTranscription` function with the `audioStream`
3. Write data to the `audioStream` from the gRPC stream as it arrives

### Transcribe the stream

```typescript
const audioStream = async function* () {
  for await (const chunk of stream) {
    const buffer = Buffer.from(chunk);
    yield { AudioEvent: { AudioChunk: buffer } };
  }
};

const command = new StartStreamTranscriptionCommand({
  LanguageCode: LanguageCode.EN_US,
  MediaEncoding: MediaEncoding.PCM,
  MediaSampleRateHertz: 16000,
  AudioStream: audioStream(),
});

const response = await client.send(command);
```

This will send the stream to Transcribe for processing. Once we get a response back from Transcribe, we will write this back to the gRPC client via the stream.

```typescript
call.write({
  message: JSON.stringify(
    event.TranscriptEvent.Transcript.Results[0].Alternatives[0].Transcript,
  ),
});
```

## gRPC Client

```typescript
async function main() {
  let target: string = 'localhost:50051';
  let credentials = grpc.credentials.createInsecure();

  if (TARGET) {
    target = `${TARGET}:50051`;
    credentials = grpc.credentials.createSsl();
  }

  console.log(`Using target: ${target}`);

  if (!WAV_FILE) {
    throw new Error('WAV_FILE required');
  }

  console.log(`Using source: ${WAV_FILE}`);

  const { audioStream } = await processFfmpeg(WAV_FILE);
}
```

The main client function will determine the `TARGET` and `WAV_FILE` to be used. This client can be used both locally or with the deployed gRPC server. A `WAV_FILE` is required and can be stored in the `client\src` directory and will be copied over during the Docker build. If the `WAV_FILE` is a stereo file with each channel a separate user, they will be muxed together when sent to the gRPC server. The result will be that the transcription is the combined result of both channels. If you want to get a single channel, you can modify the `ffmpeg` to only pass one channel, or separate the `WAV_FILE` before sending.

### ffmpeg Processing

```typescript
async function processFfmpeg(fileName: string): Promise<{
  audioStream: PassThrough;
}> {
  console.log('Processing with ffmpeg');
  const audioStream = new PassThrough();

  ffmpeg(fileName)
    .native()
    .output(audioStream)
    .format('wav')
    .audioCodec('pcm_s16le')
    .audioBitrate(8000)
    .on('error', (error: { message: string }) => {
      console.log('Cannot process: ' + error.message);
    })
    .run();

  return { audioStream };
}
```

This demo uses `ffmpeg` to ensure the audio being streamed is sent at the correct rate and with the correct formatting.

### Streaming

```typescript
const client = new audio_proto.AudioStreamer(target, credentials);
const chunkSize = 1024;
const call = client.streamAudio();

audioStream.on('data', (chunk) => {
  let offset = 0;
  while (offset < chunk.length) {
    const end = Math.min(offset + chunkSize, chunk.length);
    const chunkToSend = chunk.slice(offset, end);
    call.write({ audioData: chunkToSend });
    offset = end;
  }
});

audioStream.on('end', () => {
  console.log('Audio stream ended.');
  call.end();
});

call.on('data', (response: any) => {
  console.log(`${Date.now()}: ${response.message}`);
});

call.on('end', () => {
  console.log('gRPC call ended.');
  process.exit(0);
});

call.on('error', (err: any) => {
  console.error('Error:', err);
});
```

Once the `audioStream` has been created through `processFfmpeg`, when data arrives, it will be written to the gRPC stream in chunks through the client. Responses that are generated on the server (the transcriptions) will be surfaced through the `call.on('data', () => {})` function.

## In Action

![InActionOverview](/images/blog/node-grpc-server/InActionOverview.png)

In order to test this demo, you can either use it locally or through the deployed Fargate container.

### Deploying

```bash
yarn launch
```

This will deploy the CDK stack to your account. You will need a HostedZone domain in order to use this. The A Record created will be written to the `client\src\.env` file so that you can use the client with this gRPC server.

### Testing

In order to test locally, you will need to start the gRPC server. In the `src\resources\gRPCServer` directory, you can start the gRPC Server Docker container:

```bash
docker-compose run --build grpc-server
```

You will need to have AWS credentials exposed in this container in order to use Transcribe.

In the `client\src` directory, you can start the gRPC Client Docker container:

```bash
docker-compose run --build -e TARGET= grpc-client
```

You can add the the `WAV_FILE` environment variable in the command line as well

```bash
docker-compose run --build -e TARGET= -e WAV_FILE=EXAMPLE.wav grpc-client
```

This will use the localhost as the `TARGET`. If you want to test with the hosted gRPC server and have already deployed it, you can use:

```bash
docker-compose run --build grpc-client
```

### Client Results

![ClientResults](/images/blog/node-grpc-server//ClientResults.png)

### Server Results

![ServerResults](/images/blog/node-grpc-server/ServerResults.png)


This demo will show you how to deploy and use a gRPC server and client to stream audio and use Amazon Transcribe to analyze the audio

gRPC Streaming Audio with Node


# CDK Pipeline for ECS hosted NextJS App

In this demo we will explore and see how to use two different, but complementary techniques:

- Using CDK Pipelines to deploy a Stack
- Deploying a NextJS application to an ECS Fargate container

In each of these examples, we will look at several different mechanisms to provide security and scaling. This blog has been converted to using this method of hosting.

## GitHub Repo

https://github.com/schuettc/cdk-pipeline-ecs-hosted-nextjs

## CDK Pipeline

![CDKPipelineOverview](/images/blog/cdk-pipeline-ecs-hosted-nextjs/CDKPipelineOverview.png)

This demo uses [AWS CodeStar Connections](https://docs.aws.amazon.com/dtconsole/latest/userguide/connections-create-github.html) to create a link to a GitHub repo. This will allow us to use GitHub commits to push changes to our NextJS App through the CDK Pipeline.

The [cdk-pipeline-ecs-hosted-nextjs.ts](https://github.com/schuettc/cdk-pipeline-ecs-hosted-nextjs/blob/main/src/cdk-pipeline-ecs-hosted-nextjs.ts) file will create a Stack with a self-mutating CodePipeline. This is the Pipeline contains a Stage that will deploy the the NextJSApp Stack. This Pipeline is triggered by the GitHub repo and branch associated.

```typescript
const pipeline = new CodePipeline(this, 'Pipeline', {
  synth: new CodeBuildStep('Synth', {
    input: CodePipelineSource.connection(props.githubRepo, props.githubBranch, {
      connectionArn: props.connectionArn,
    }),
    commands: ['yarn install --frozen-lockfile', 'yarn build', 'npx cdk synth'],
    rolePolicyStatements: [
      new PolicyStatement({
        actions: ['sts:AssumeRole'],
        resources: ['*'],
        conditions: {
          StringEquals: {
            'iam:ResourceTag/aws-cdk:bootstrap-role': 'lookup',
          },
        },
      }),
    ],
  }),
  selfMutation: true,
  dockerEnabledForSynth: true,
});

pipeline.addStage(
  new NextJSAppStage(this, 'NextJSApp', props as NextJSAppProps),
);
```

## NextJS App

![NextJSAppOverview](/images/blog/cdk-pipeline-ecs-hosted-nextjs/ECSFargateNextJSApp.png)

The demo optionally uses a pre-configured Route 53 Hosted Zone domain. Otherwise the traffic will route through the following services:

- Route 53 (Optional)
- Cloudfront
- Application Load Balancer
- Fargate with Auto Scaling Group

### Securing Traffic

In order to ensure that traffic uses this path, we will establish the follow security mechanisms.

#### Fargate

```typescript
const fargateTargetGroup = new ApplicationTargetGroup(
  this,
  'nextJsAppTargetGroup',
  {
    vpc: props.vpc,
    port: 3000,
    protocol: ApplicationProtocol.HTTP,
    targets: [this.fargateService],
    healthCheck: {
      path: '/',
      protocol: Protocol.HTTP,
      port: '3000',
    },
  },
);
```

When configuring the `ApplicationTargetGroup` that points to the Fargate application, a Security Group will be created that allows traffic from the Application Load Balancer to access the Fargate task.

![FargateSecurityGroup](/images/blog/cdk-pipeline-ecs-hosted-nextjs/FargateSecurityGroup.png)

This Security Group will use the Security Group of the Application Load Balancer.

### Application Load Balancer

```typescript
this.applicationLoadBalancerSecurityGroup.addIngressRule(
  Peer.anyIpv4(),
  Port.tcp(80),
);
```

This Security Group will allow traffic in from any IPv4 address on TCP Port 80. In order to ensure that traffic from sources besides Cloudfront are rejected, we will use a configuration of the Listener.

```typescript
const fargateListener = this.applicationLoadBalancer.addListener(
  'fargateListener',
  {
    port: 80,
    protocol: ApplicationProtocol.HTTP,
    open: true,
    defaultAction: ListenerAction.fixedResponse(403),
  },
);

fargateListener.addAction('ForwardFromCloudFront', {
  action: ListenerAction.forward([fargateTargetGroup]),
  conditions: [
    ListenerCondition.httpHeader(props.customHeader, [props.randomString]),
  ],
  priority: 1,
});
```

This Listener will by default reject all traffic that is sent to it. An additional action will be used to allow traffic with a specific header and value to forward to the Fargate application. This information is created separately and configured during deployment. For more information on this mechanism, see [building-a-websocket-server-with-fargate-and-cdk](https://subaud.io/blog/building-a-websocket-server-with-fargate-and-cdk).

The result:

![Listener](/images/blog/cdk-pipeline-ecs-hosted-nextjs/Listener.png)

### Cloudfront

```typescript
this.distribution = new Distribution(this, 'CloudfrontDistribution', {
  minimumProtocolVersion: SecurityPolicyProtocol.TLS_V1_2_2021,
  defaultBehavior: {
    origin: new LoadBalancerV2Origin(props.applicationLoadBalancer, {
      httpPort: 80,
      protocolPolicy: OriginProtocolPolicy.HTTP_ONLY,
      originId: 'default-origin',
    }),
    viewerProtocolPolicy: ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
    cachePolicy: CachePolicy.CACHING_DISABLED,
    allowedMethods: AllowedMethods.ALLOW_GET_HEAD_OPTIONS,
    originRequestPolicy: OriginRequestPolicy.ALL_VIEWER,
  },
  domainNames: props.domainName ? [props.domainName] : undefined,
  certificate: certificate,
  priceClass: PriceClass.PRICE_CLASS_100,
});
```

The Distribution created includes a domain and SSL certificate to provide HTTPS access to the Fargate task. Requests to HTTP will redirect to HTTPS.

#### Domain (optional)

```typescript
if (props.domainName) {
  hostedZone = HostedZone.fromLookup(this, 'HostedZone', {
    domainName: props.domainName,
  });

  certificate = new Certificate(this, 'Certificate', {
    domainName: props.domainName,
    validation: CertificateValidation.fromDns(hostedZone),
  });
}

if (hostedZone && props.domainName) {
  new AaaaRecord(this, 'mainAaaaRecord', {
    zone: hostedZone,
    target: RecordTarget.fromAlias(new CloudFrontTarget(this.distribution)),
  });

  new ARecord(this, 'mainARecord', {
    zone: hostedZone,
    target: RecordTarget.fromAlias(new CloudFrontTarget(this.distribution)),
  });
}
```

If a domain is included in the deployment, a certificate will be created and used in addition to the default Cloudfront certificate. DNS records will be created associated with this domain.

#### Custom Header

To create the header used to ensure traffic is sourced from Cloudfront, the Cloudfront Distribution Origin will be configured to send it using a Custom Resource.

```typescript
new CustomResource(this, 'customHeaderCustomResource', {
  serviceToken: customHeaderCustomResourceProvider.serviceToken,
  properties: {
    DistributionId: this.distribution.distributionId,
    Origins: [
      {
        OriginId: 'default-origin',
        CustomHeaders: [
          {
            HeaderName: props.customHeader,
            HeaderValue: props.randomString,
          },
        ],
      },
    ],
  },
});
```

The result will look something like this:

![Origin](/images/blog/cdk-pipeline-ecs-hosted-nextjs/Origin.png)

### Fargate Task

```typescript
const ecsTask = new FargateTaskDefinition(this, 'ecsTask', {
  memoryLimitMiB: 2048,
  cpu: 1024,
  runtimePlatform: {
    operatingSystemFamily: OperatingSystemFamily.LINUX,
    cpuArchitecture: CpuArchitecture.X86_64,
  },
});

ecsTask.addContainer('NextJsContainer', {
  image: ContainerImage.fromAsset('nextjs-docker'),
  containerName: 'nextJsApp',
  portMappings: [{ containerPort: 3000, hostPort: 3000 }],
  logging: LogDrivers.awsLogs({
    streamPrefix: 'nextJsApp',
  }),
  environment: {},
});

const taskSecurityGroup = new SecurityGroup(this, 'taskSecurityGroups', {
  vpc: props.vpc,
});

this.fargateService = new FargateService(this, 'nextJsAppFargateService', {
  cluster: cluster,
  taskDefinition: ecsTask,
  assignPublicIp: true,
  desiredCount: 1,
  vpcSubnets: { subnetType: SubnetType.PUBLIC },
  securityGroups: [taskSecurityGroup],
});
```

The Fargate Service will be configured with a port mapping for port 3000. This corresponds to the Application Load Balancer that forwards traffic to port 3000. The Fargate task will be deployed in the VPC created during the deployment and assigned a public IP address.

The Docker image used is the example shown here: https://github.com/vercel/next.js/tree/canary/examples/with-docker

### Result

The result is an Application Load Balancer that will only accept traffic with a specific header and value. This header and value are known only by the deployed Cloudfront Distribution. Traffic with this header will be forwarded to a Fargate task that only allows traffic from the Security Group associated with the Application Load Balancer. This deployment allows us to restrict access to the the Fargate task that uses the Cloudfront Distribution.

## Configuration and Deployment

To deploy this repo, you will need to complete the following steps:

1. Fork this repo
2. Configure Codestar connections with your GitHub repo
3. Configure a Hosted Zone in Route 53 with your domain (optional)
4. Configure your `.env` file
5. Deploy

### Configure Codestar

To configure your AWS account to support a connection to GitHub, follow the instructions here: https://docs.aws.amazon.com/dtconsole/latest/userguide/connections-create-github.html

The result will look something like this:

![Connections](/images/blog/cdk-pipeline-ecs-hosted-nextjs/Connections.png)

### Configure Hosted Zone (optional)

To create a Hosted Zone, follow the instructions here: https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/CreatingHostedZone.html

The result will look something like this:

![HostedZone](/images/blog/cdk-pipeline-ecs-hosted-nextjs/HostedZone.png)

### Configure `.env` file

Before deploying, you will need to configure a `.env` file with the following fields:

```bash
CONNECTION_ARN=arn:aws:codestar-connections:us-east-1:112233445566:connection/XXXXXXXXXXXXXXXXXXXXXXXX
GITHUB_REPO=XXXXXXXXXXX/cdk-pipeline-ecs-hosted-nextjs
GITHUB_BRANCH=main
DOMAIN_NAME=example.com
```

`DOMAIN_NAME` is optional

### Deploy

To deploy the Pipeline:

```bash
yarn launch
```

Deploying the Pipeline will begin the process of deploying the NextJSApp stack.


A CDK Pipeline example that will deploy a NextJS App on ECS

CDK Pipeline for ECS hosted NextJS App


# ffmpeg streaming to KVS

[GitHub Repo](https://github.com/schuettc/ffmpeg-kvs-streaming)

[Amazon Kinesis Video Streams](https://aws.amazon.com/kinesis/video-streams) can be used to get access to other AWS services with streaming video. In this demo, we will see how to use [Node.js](https://nodejs.org/en) and [ffmpeg](https://ffmpeg.org/) to convert a file into a stream that can be sent to KVS.

![Overview](/images/blog/streaming-to-kvs-with-ffmpeg/ffmpegStreaming.png)

1. An object is created in the Amazon Simple Storage Service (Amazon S3) bucket. When this happens, a notification is sent to the associated AWS Lambda function.
2. This Lambda makes a request to the [Application Load Balancer](https://docs.aws.amazon.com/elasticloadbalancing/latest/application/introduction.html) associated with the [AWS Fargate](https://aws.amazon.com/fargate/) task with the object information.
3. The Fargate application downloads the object from S3 and begins processing (processing details below).
4. The Fargate application streams the contents of the object to [Amazon Kinesis Video Streams](https://aws.amazon.com/kinesis/video-streams/).

## Processing with ffmpeg

Once the file has been downloaded from S3, the application will begin processing the file with ffmpeg.

```typescript
async function processFfmpeg(fileName: string): Promise<{
  videoStream: PassThrough;
}> {
  console.log('Processing with ffmpeg');
  const videoStream = new PassThrough();
  let videStreamCount: number = 0;

  ffmpeg(fileName)
    .native()
    .output(videoStream)
    .outputOptions(['-profile:v baseline'])
    .videoCodec('libx264')
    .size('640x480')
    .audioCodec('aac')
    .format('matroska')
    .on('error', (error) => {
      console.log('Cannot process: ' + error.message);
    })
    .on('stderr', (data) => {
      if (videStreamCount % COUNT_FREQUENCY === 0) {
        console.info(`videoStream: ${data}`);
      }
      videStreamCount++;
    })
    .run();

  return { videoStream };
}
```

To do this, we will create a `PassThrough` stream that will be used as the output of the ffmpeg pipe. By using `.native()` as an input option, the ffmpeg pipe will read the contents of the file at the native frame rate. The output options used will format the output stream so that it can be used by KVS.

## Streaming Media

Because Kinesis Video Streams [PutMedia](https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/API_dataplane_PutMedia.html) does not have a corresponding AWS SDK command, we must use a direct HTTP request. This involves several extra steps:

1. GetDataEndpoint
2. GetCredentials
3. SignRequest
4. PutMedia

### GetDataEndpoint

```typescript
async function getEndpoint(
  streamArn: string,
): Promise<GetDataEndpointCommandOutput['DataEndpoint']> {
  const response = await kvsClient.send(
    new GetDataEndpointCommand({
      APIName: APIName.PUT_MEDIA,
      StreamARN: streamArn,
    }),
  );
  return response.DataEndpoint;
}
```

The first step to sending a stream to KVS is to get an endpoint for the stream we are going to use. We have already created the stream during the deployment of the CDK, so we will use that stream as the target.

### GetCredentials

```typescript
const ECS_ROLE = process.env.ECS_ROLE || '';

async function getCredentials(): Promise<
  AssumeRoleCommandOutput['Credentials']
> {
  console.log('Getting credentials');
  const response = await stsClient.send(
    new AssumeRoleCommand({
      RoleArn: ECS_ROLE,
      RoleSessionName: 'kvs-stream',
    }),
  );
  return response.Credentials;
}
```

In order to stream to the KVS, we will need to assume an IAM role that has permission to do this. More details about doing this using a local Docker are outlined below.

#### IAM Role

```typescript
this.kinesisRole = new Role(this, 'kinesisRole', {
  assumedBy: new CompositePrincipal(
    new ArnPrincipal(kvsProducerRole.roleArn),
    new AccountPrincipal(Stack.of(this).account),
  ),
  inlinePolicies: {
    ['KinesisVideoPolicy']: new PolicyDocument({
      statements: [
        new PolicyStatement({
          resources: ['*'],
          actions: ['kinesisvideo:PutMedia'],
        }),
      ],
    }),
  },
});
```

During the deployment of the CDK, an [IAM Role](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) is created. This is the role that we will assume and use to `PutMedia`. This Role is passed to the application through environment variables.

### SignRequest

```typescript
async function signRequest(
  streamArn: string,
  endpoint: string,
): Promise<{ signedUrl: aws4.Request; reqUrl: string }> {
  const credentials = await getCredentials();
  if (!credentials) {
    throw new Error('Failed to get credentials');
  }
  const signedUrl = aws4.sign(
    {
      host: endpoint,
      path: '/putMedia',
      service: 'kinesisvideo',
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-amzn-stream-arn': streamArn,
        'x-amzn-fragment-timecode-type': 'ABSOLUTE',
        'x-amz-content-sha256': 'UNSIGNED-PAYLOAD',
      },
    },
    {
      accessKeyId: credentials.AccessKeyId,
      secretAccessKey: credentials.SecretAccessKey,
      sessionToken: credentials.SessionToken,
    },
  );
  if (!signedUrl || !signedUrl.host || !signedUrl.path) {
    throw new Error('Failed to sign request');
  }
  const reqUrl = signedUrl.host + signedUrl.path;
  return { signedUrl: signedUrl, reqUrl: reqUrl };
}
```

Using the credentials from the assumed role, the Stream ARN, and the DataEndpoint, we will use [aws4](https://github.com/mhart/aws4) to sign the request we will be making and get a URL to use in the `PutMedia`.

### PutMedia

```typescript
const agent = new Agent({
  rejectUnauthorized: false,
});
const axiosParams = {
  method: 'POST',
  timeout: 40 * 1000,
  url: reqUrl,
  headers: signedUrl.headers as AxiosHeaders,
  data: stream,
  responseType: 'stream' as ResponseType,
  maxContentLength: Infinity,
  httpsAgent: agent,
};
```

Next, we will combine all of the pieces together as we prepare to `PutMedia`. Using the stream generated by `ffmpeg` and the url generated by `aws4`, we will create a `POST` request using [axios](https://axios-http.com/docs/intro).

```typescript
let streamCount: number = 0;

try {
  const response = (await axios(axiosParams)) as AxiosResponse;
  const startFragmentNumber = await getStartFragmentNumber(response);

  response.data.on('data', (chunk: Buffer) => {
    if (streamCount % COUNT_FREQUENCY === 0) {
      console.info(`Stream: ${streamArn} - Chunk: ${chunk.toString()}`);
    }
    streamCount++;
  });
  response.data.on('end', async () => {
    console.log(`Stream: ${streamArn} ended`);
  });

  response.data.on('error', (error: Error) => {
    console.error(`Error in ${streamArn} stream: ${error}`);
  });
  if (startFragmentNumber) {
    return { streamArn, startFragmentNumber };
  } else {
    throw new Error('StartFragment not found');
  }
} catch (error) {
  console.error('Error in putMedia:', error);
  throw error;
}
```

Finally, we will make the `POST` request to the `PutMedia` API and process the responses. For every fragment sent, KVS will acknowledge with a response:

```json
{
       Acknowledgement : {
          "EventType": enum
          "FragmentTimecode": Long,
          "FragmentNumber": Long,
          "ErrorId" : String
      }
}
```

## Testing

In order to test this demo, you can upload a movie file to the S3 bucket that is created during the CDK deployment. This will start processing on the Fargate container and streaming to KVS. This stream can be seen in the [KVS Console](https://us-east-1.console.aws.amazon.com/kinesisvideo/home?region=us-east-1#/streams) and finding the stream named `ffmpeg-streaming`.

![KVSConsole](/images/blog/streaming-to-kvs-with-ffmpeg/KVS.png)

This stream can be viewed with `Media playback`.

![MediaPlayback](/images/blog/streaming-to-kvs-with-ffmpeg/MediaPlayback.png)

## Developing Locally with Docker

Because development like this can require extensive trial and error testing, it is useful to test locally with a Docker image before deploying to Fargate. This will involve several steps:

1. Build a local Docker container
2. Pass credentials and environment variables to this container
3. Run the local Docker container
4. Trigger this container

### Build the Docker container

To do this, we will use [Docker Compose](https://docs.docker.com/compose/) and a `.env` file.

```docker-compose
version: '3'
services:
  ffmpeg:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - AWS_PROFILE
      - AWS_ACCESS_KEY_ID
      - AWS_SECRET_ACCESS_KEY
      - AWS_SESSION_TOKEN
      - COUNT_FREQUENCY=60
      - ECS_ROLE
      - KVS_STREAM_ARN
    volumes:
      - ~/.aws/:/root/.aws:ro
    ports:
      - '80:80'
    command: ['npm', 'start']
```

### Pass credentials and environment variables

This will use the `Dockerfile` used in the deployment with environment variables and a mounted volume that references `~./aws`. This will allow us to pass credentials from the [`~.aws/credentials`](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html) file and use the associated `AWS_PROFILE`.

Alternatively, you can export `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_SESSION_TOKEN` as environment variables.

`ECS_ROLE` and `KVS_STREAM_ARN` are created during the deployment of the CDK and passed to the `.env` file through the outputs of the CDK.

### Run the Docker container

To run the Docker container:

```bash
docker-compose up --build
```

This will re-build and run the Docker container

### Trigger the Docker container

[Postman](https://www.postman.com/) is a useful tool for making API request. This is similar to what we will be doing with the Lambda function and serves as a convenient way to test locally. The Docker container will be running on port 80, so we will make a `POST` request to `localhost:80/processObject` passing the `bucketName` and `keyName` of the object to be processed.

![Postman](/images/blog/streaming-to-kvs-with-ffmpeg/Postman.png)

## Using this demo

### Requirements

- yarn installed
- ARM processor
- Docker desktop running

### Deploy

```bash
yarn launch
```

### Cleanup

```bash
yarn cdk destroy
```


This demo will go through the steps to process a file with ffmpeg and stream the contents to KVS using Node.js

Streaming to KVS with ffmpeg and Node.js


[Feature flags](https://docs.aws.amazon.com/cdk/v2/guide/featureflags.html) are a way to modify the deployment behavior of an AWS SDK deployment. These are potentially breaking, so use with caution. These feature flags are stored in the [context](https://docs.aws.amazon.com/cdk/v2/guide/context.html) of the CDK as key-value pairs.

For example:

```json
  "context": {
    "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true
  },
```

Here we are setting a feature flag to tell the CDK to always use a unique launch template name. This is useful when [requiring imdsv2](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_ec2-readme.html#configuring-instance-metadata-service-imds) for an EC2 instance.

```typescript
    const ec2Instance = new Instance(this, 'Instance', {
      vpc: props.vpc,
      instanceType: InstanceType.of(InstanceClass.C7G, InstanceSize.MEDIUM),
      machineImage: ubuntuAmi,
      requireImdsv2: true,
      userData: userData,
      ...
    })
```

Without this feature flag enabled, you are likely to get an error similar to this:

```bash
Launch template name already in use. (Service: AmazonEC2; Status Code: 400; Error Code: InvalidLaunchTemplateName.AlreadyExistsException;
```

To add this feature flag to the `cdk.json` file when using [projen](https://projen.io/) we must modify the `.projenrc.ts` file.

```typescript
project.cdkConfig.json.addOverride('context', {
  '@aws-cdk/aws-ec2:uniqueImdsv2TemplateName': true,
});
```

This will add the necessary context to the `cdk.json` file and enable the [`uniqueImdsv2TemplateName`](https://github.com/aws/aws-cdk/blob/main/packages/%40aws-cdk/cx-api/FEATURE_FLAGS.md#aws-cdkaws-ec2uniqueimdsv2templatename) feature flag.

For a complete list of available feature flags available: https://github.com/aws/aws-cdk/blob/main/packages/%40aws-cdk/cx-api/FEATURE_FLAGS.md


Feature flags can be used to modify the deployment characteristics of a CDK deployment.  These feature flags can be enabled in projen by modifying the cdk.json file.

Adding feature flags to AWS CDK through runtime context with projen


# CDK Deployment of Docker Image with CodeBuild

In many CDK deployments, a local Docker build can be done to create assets. For example, when deploying a [Python Lambda with Requirements](https://subaud.io/blog/deploying-python-lambda-with-requirements-using-cdk) you can use a [bundling](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_s3_assets.AssetOptions.html#bundling) to install dependencies before the Function is zipped and uploaded. This can cause issues if the CPU of the machine doing the deploy is different from the [Architecture](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_lambda.Architecture.html) of the [Function](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_lambda.Function.html). If the CPU of the machine doing the deploy is an Intel based machine, but the Function is `ARM_64` the Function will not work. The same applies to [Fargate](https://docs.aws.amazon.com/AmazonECS/latest/userguide/what-is-fargate.html) containers. In this demo, you will see how to deploy a Fargate container without having to worry about the CPU of the deploying machine using [AWS CodeBuild](https://aws.amazon.com/codebuild/).

## Goal

In this demo, we are trying to build a CDK deployed Fargate container that is accessed through a Cloudfront Distribution and Application LoadBalancer using a custom Docker build.

![GoalOverview](/images/blog/CodeBuildOverviewGoal.png)

In order to do this, we must build a [Docker image](https://docs.aws.amazon.com/AmazonECS/latest/userguide/create-container-image.html). We want to use an `ARM_64` cpuArchitecture for this image, but do not want to require the CDK to be deployed from a machine with an `ARM_64` CPU. Therefore, we will use CodeBuild to build the image.

## ECS Setup

The first step to deploying this image is to create the ECS infrastructure. Much of this is covered [here](https://www.subaud.io/blog/building-a-websocket-server-with-fargate-and-cdk) in more detail.

The biggest difference is that we will initialize this ECS cluster with a container that we will not be using for long.

```typescript
taskDefinition.addContainer('cdk-codebuild', {
  image: ContainerImage.fromRegistry(
    'public.ecr.aws/nginx/nginx:latest-arm64v8',
  ),
  portMappings: [{ containerPort: 80, hostPort: 80 }],
  logging: new AwsLogDriver({ streamPrefix: 'cdk-codebuild' }),
  healthCheck: {
    command: ['CMD-SHELL', 'curl -f http://localhost/'],
  },
});
```

The container that we are adding here is only a placeholder until we can build and deploy our actual image. To do this, we will use an `ARM_64` [nginx](https://gallery.ecr.aws/nginx/nginx) image. When the ECS service is created, this image will be used for the Task created.

## Building the Image

![BuildOverview](/images/blog/CodeBuildOverviewBuild.png)

Uploading and building the image is a multi-step process that will take local code and deploy it to our Fargate Service.

### Uploading Assets

The first step to deploying this image is to upload the local `Dockerfile` and associated code.

```typescript
const bundle = new Asset(this, 'bundle', {
  path: 'src/resources/dockerExample',
});
```

This will zip the contents of the `dockerExample` directory and upload to the CDK assets bucket.

### Pipeline

In order to build the image, we will use an [AWS CodePipeline](https://aws.amazon.com/codepipeline/) with this bucket and key as the source.

```typescript
const sourceAction = new S3SourceAction({
  actionName: 'S3Source',
  bucket: bundle.bucket,
  bucketKey: bundle.s3ObjectKey,
  output: sourceOutput,
  trigger: S3Trigger.EVENTS,
});
```

Now, whenever this key is updated, the AWS CodePipeline will trigger. This pipeline is only three steps - source, build, and deploy:

```typescript
new Pipeline(this, 'docker-codebuild-pipeline', {
  stages: [
    {
      stageName: 'source',
      actions: [sourceAction],
    },
    {
      stageName: 'build',
      actions: [buildAction],
    },
    {
      stageName: 'deploy',
      actions: [deployAction],
    },
  ],
});
```

### CodeBuild

The build portion of this pipeline is the most complex but can be broken down to a few key steps:

1. Using the source key (the uploaded Zip file)
2. A `docker build` command is executed
3. This image is pushed to an ECR repository
4. An `imagedefinitions.json` file is created with this information

```typescript
const project = new Project(this, 'codeBuildProject', {
  role: codeBuildRole,
  buildSpec: BuildSpec.fromObject({
    version: '0.2',
    phases: {
      pre_build: {
        commands: [
          'echo Logging in to Amazon ECR...',
          'aws ecr get-login-password --region $AWS_DEFAULT_REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com',
        ],
      },
      build: {
        commands: [
          'echo Build started on `date`',
          'echo Building the Docker image...',
          'docker build -t $IMAGE_REPO_NAME:$IMAGE_TAG .',
          'docker tag $IMAGE_REPO_NAME:$IMAGE_TAG $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com/$IMAGE_REPO_NAME:$IMAGE_TAG',
        ],
      },
      post_build: {
        commands: [
          'echo Build completed on `date`',
          'echo Pushing the Docker image...',
          'docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com/$IMAGE_REPO_NAME:$IMAGE_TAG',
          'echo Writing image definitions file...',
          'printf \'[{"name":"cdk-codebuild","imageUri":"%s"}]\' $IMAGE_REPO_URI:$IMAGE_TAG > imagedefinitions.json',
        ],
      },
    },
    artifacts: {
      files: ['imagedefinitions.json'],
    },
  }),
  source: Source.s3({
    bucket: bundle.bucket,
    path: bundle.s3ObjectKey,
  }),
  environment: {
    buildImage: LinuxArmBuildImage.AMAZON_LINUX_2_STANDARD_3_0,
    environmentVariables: {
      AWS_DEFAULT_REGION: { value: Stack.of(this).region },
      AWS_ACCOUNT_ID: { value: Stack.of(this).account },
      IMAGE_REPO_NAME: { value: this.ecrRepository.repositoryName },
      IMAGE_REPO_URI: { value: this.ecrRepository.repositoryUri },
      IMAGE_TAG: { value: 'latest' },
    },
  },
});
```

This will use a `LinuxArmBuildImage` to create the Docker image, ensuring us that an `ARM_64` CPU will be used when the image is created regardless of the CPU of the machine doing the CDK deployment.

### Deployment

Finally, we will use this image to deploy to the Fargate service previously created.

```typescript
const deployAction = new EcsDeployAction({
  actionName: 'Deploy',
  service: props.fargateService,
  input: buildOutput,
});
```

The `public.ecr.aws/nginx/nginx:latest-arm64v8` image that was originally used will be replaced with the newly built image that has been uploaded to an ECR repository.

## Caveats

In a deployment like this, the CodePipeline will only be triggered when the source code for the Docker image is changed. If you need to rebuild the image for any reason, you will need to manually run the CodePipeline.

## GitHub Repo

This demo is available here: https://github.com/schuettc/cdk-docker-codebuild

## Testing

To test this for yourself, simply clone this repo and deploy the CDK:

```bash
git clone git@github.com:schuettc/cdk-docker-codebuild.git
cd cdk-docker-codebuild
yarn launch
```

## Cleanup

To remove these resources:

```bash
yarn cdk destroy
```


In this demo we will see how to deploy a custom built Fargate container using CodeBuild

CDK Deployment of Docker Image with CodeBuild

# Building a Private Fargate Task and Lambda

In order to build an AWS Fargate task in a [private subnet](https://docs.aws.amazon.com/vpc/latest/userguide/configure-subnets.html) that can still be invoked by a Lambda function, the Lambda function must also be built in the same VPC with access to the private subnet.

![Overview](/images/blog/PrivateLambdaFargate.png)

## GitHub Repo

Full code is available here: https://github.com/schuettc/cdk-private-fargate-with-lambda

## Building the VPC and ALB

```typescript
this.vpc = new Vpc(this, 'VPC', {
  subnetConfiguration: [
    {
      cidrMask: 24,
      name: 'PrivateSubnet',
      subnetType: SubnetType.PRIVATE_WITH_EGRESS,
    },
    {
      cidrMask: 24,
      name: 'PublicSubnet',
      subnetType: SubnetType.PUBLIC,
    },
  ],
  maxAzs: 2,
});
```

Here we are creating a VPC with two types of Subnets - `PRIVATE_WITH_EGRESS` and `PUBLIC`.

```typescript
this.applicationLoadBalancer = new ApplicationLoadBalancer(this, 'alb', {
  vpc: this.vpc,
  vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_EGRESS },
  internetFacing: false,
  securityGroup: this.fargateSecurityGroup,
});
```

Next we will create an Application Load Balancer in the `PRIVATE_WITH_EGRESS` subnet.

## Fargate

```typescript
this.fargateService = new ApplicationLoadBalancedFargateService(
  this,
  'fargateService',
  {
    taskImageOptions: {
      image: ContainerImage.fromRegistry('amazon/amazon-ecs-sample'),
      taskRole: fargateTaskRole,
    },
    publicLoadBalancer: false,
    vpc: props.vpc,
    assignPublicIp: false,
    openListener: false,
    loadBalancer: props.applicationLoadBalancer,
    listenerPort: 80,
    taskSubnets: {
      subnetType: SubnetType.PRIVATE_WITH_EGRESS,
    },
    securityGroups: [props.fargateSecurityGroup],
    runtimePlatform: {
      operatingSystemFamily: OperatingSystemFamily.LINUX,
      cpuArchitecture: CpuArchitecture.X86_64,
    },
  },
);
```

Next we will create the Fargate Service and Task, once again assigning it to the `PRIVATE_WITH_EGRESS` subnet using the previously created Application Load Balancer.

## Lambda

```typescript
const fargateLambdaInPrivateVPCwithSG = new NodejsFunction(
  this,
  'fargateLambdaInPrivateVPCwithSG',
  {
    entry: 'src/resources/fargateLambda/index.ts',
    runtime: Runtime.NODEJS_LATEST,
    architecture: Architecture.ARM_64,
    role: lambdaRole,
    timeout: Duration.seconds(60),
    vpc: props.vpc,
    vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_EGRESS },
    securityGroups: [lambdaSecurityGroup],
    environment: {
      FARGATE_ALB_URL: props.fargateService.loadBalancer.loadBalancerDnsName,
    },
  },
);
```

Finally, we will create the Lambda function and assign to the `PRIVATE_WITH_EGRESS` subnet.

## Security Groups

In addition to containing the Lambda, Application Load Balancer, and Fargate within the same VPC and private subnet, Security Groups can also be used to restrict access.

```typescript
this.fargateSecurityGroup = new SecurityGroup(this, 'fargateSecurityGroup', {
  vpc: this.vpc,
  description: 'Security Group for Fargate ALB',
});
```

First we will create a Security Group for the Application Load Balancer.

```typescript
const lambdaSecurityGroup = new SecurityGroup(this, 'LambdaSecurityGroup', {
  vpc: props.vpc,
});
```

Next we will create a Security Group for the Lambda function.

```typescript
props.fargateSecurityGroup.connections.allowFrom(
  new Connections({
    securityGroups: [lambdaSecurityGroup],
  }),
  Port.tcp(80),
  'allow traffic on port 80 from the Lambda security group',
);
```

Finally, we will allow a connection from the Lambda Security Group to the Application Load Balancer Security Group. This will allow the Lambda function to make requests to the Fargate Task through the Application Load Balancer. The other two Lambda functions show how this is required as they do not work. The first does not include the correct Security Group and the second does not include the correct VPC subnet.

## Result

Once deployed, the Lambda function in the correct VPC with the correct Security Group will be able to make a request to the Fargate task and get a response:

```html
<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1" />
    <title>Simple PHP App</title>
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <link href="assets/css/bootstrap.min.css" rel="stylesheet" />
    <style>
      body {
        margin-top: 40px;
        background-color: #333;
      }
    </style>
    <link href="assets/css/bootstrap-responsive.min.css" rel="stylesheet" />
    
  </head>

  <body>
    <div class="container">
      <div class="hero-unit">
        <h1>Simple PHP App</h1>
        <h2>Congratulations</h2>
        <p>Your PHP application is now running on a container in Amazon ECS.</p>
        <p>The container is running PHP version 5.4.16.</p>
      </div>
    </div>
  </body>
</html>
```

The other Lambda functions will timeout and fail

```text
Task timed out after 60.07 seconds
```

Additionally, the Application Load Balancer URL will only return an RFC 1918 private address:

```bash
$ nslookup internal-Privat-vpcRe-8uesdCpJs6j6-1716194524.us-east-1.elb.amazonaws.com

Non-authoritative answer:
Name:	internal-privat-vpcre-8uesdcpjs6j6-1716194524.us-east-1.elb.amazonaws.com
Address: 10.0.1.253
Name:	internal-privat-vpcre-8uesdcpjs6j6-1716194524.us-east-1.elb.amazonaws.com
Address: 10.0.0.230
```

How to build a private Fargate task and connect to it with an AWS Lambda function using AWS CDK.  

All posts

gRPC Streaming Audio with Node

CDK Pipeline for ECS hosted NextJS App

Streaming to KVS with ffmpeg and Node.js

Adding feature flags to AWS CDK through runtime context with projen

CDK Deployment of Docker Image with CodeBuild

Using CDK to build a Fargate task with AWS Lambda integration