DEV Community: Randy Findley

Live Streaming Origin

Randy Findley — Tue, 16 Jun 2020 11:53:08 +0000

If you're just joining us, please take a look at part 1 in this 3-part series.

Part 1 - Server
Part 2 - Proxy
Part 3 - Origin

This blog post is about the Origin. Remember we have 3 services in this architecture. Proxy -> Server <- Origin

How do we route & cache HTTP traffic to a fleet of RTMP Servers to serve the HLS files?

We need a single endpoint like live.finbits.io.

A user will play the the video using a URL like live.finbits.io/<stream key>/live.m3u8.

HTTP is stateless so we can use an AWS ALB load balancer. Yay!

We also use AWS CloudFront as the CDN. It looks like this.

Route 53 -> CloudFront -> ALB -> Origin(s) -> Server(s).

But how does the Origin know which Server to fetch the HLS files from?

We use a Redis cache to store the mapping between "stream key" and Server "IP:PORT".

Our Origin is simply NGINX with a small backed that performs the Redis cache lookup.

Let's take a look at our NGINX config.

worker_processes  auto;

error_log /dev/stdout info;

events {
  worker_connections  1024;
}

http {
  include       /etc/nginx/mime.types;
  default_type  application/octet-stream;

  log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                    '$status $body_bytes_sent "$http_referer" '
                    '"$http_user_agent" "$http_x_forwarded_for"';

  access_log /dev/stdout main;

  sendfile        on;

  keepalive_timeout  65;

  gzip on;

  proxy_cache_path /tmp/cache/ levels=1:2 keys_zone=CONTENTCACHE:10m max_size=15g inactive=10m use_temp_path=off;

  ignore_invalid_headers off;

  upstream node-backend {
    server localhost:3000 max_fails=0;
  }

  <% servers.forEach(function(server, index) { %>
  upstream media<%= index %>-backend {
    server <%= server %> max_fails=0;
  }
  <% }); %>

  server {
    listen 80;
    server_name localhost;
    sendfile off;

    <% servers.forEach(function(server, index) { %>
    location ~ ^/<%= server %>/(.*)$ {
      internal;
      proxy_pass http://media<%= index %>-backend/$1$is_args$args;
    }
    <% }); %>

    location ~ ^/(.*live\.m3u8)$ {
      #
      # Cache results on local disc
      #
      proxy_cache CONTENTCACHE;
      proxy_cache_lock on;
      proxy_cache_key $scheme$proxy_host$uri;
      proxy_cache_valid 1m;
      proxy_cache_use_stale error timeout invalid_header updating http_500 http_502 http_503 http_504;

      #
      # CORS
      #
      include /etc/nginx/nginx.cors.conf;

      #
      # Proxy Pass
      #
      proxy_pass http://node-backend/$1$is_args$args;
    }

    location ~ ^/(.*index\.m3u8)$ {
      #
      # Cache results on local disc
      #
      proxy_cache CONTENTCACHE;
      proxy_cache_lock on;
      proxy_cache_key $scheme$proxy_host$uri;
      proxy_cache_valid 1s;
      proxy_cache_use_stale error timeout invalid_header updating http_500 http_502 http_503 http_504;

      #
      # CORS
      #
      include /etc/nginx/nginx.cors.conf;

      #
      # Proxy Pass
      #
      proxy_pass http://node-backend/$1$is_args$args;
    }

    location ~ ^/(.*\.ts)$ {
      #
      # Cache results on local disc
      #
      proxy_cache CONTENTCACHE;
      proxy_cache_lock on;
      proxy_cache_key $scheme$proxy_host$uri;
      proxy_cache_valid 60s;
      proxy_cache_use_stale error timeout invalid_header updating http_500 http_502 http_503 http_504;

      #
      # CORS
      #
      include /etc/nginx/nginx.cors.conf;

      #
      # Proxy Pass
      #
      proxy_pass http://node-backend/$1$is_args$args;
    }

    location /healthcheck {
      proxy_pass http://node-backend/healthcheck$is_args$args;
    }

    location /nginx_status {
      stub_status on;

      access_log off;
      allow 127.0.0.1;
      deny all;
    }
  }
}

When a request comes in, like live.finbits.io/<stream key>/live.m3u8, it first hits the node-backend.

The node-backend performs the Redis cache lookup, to get the Server IP:PORT, then responds with an internal NGINX redirect to the corresponding media-backend.

The media-backend performs a proxy_pass to the Server to fetch the HLS.

What about caching?

The cache headers are added by the Server.

The Origin has an internal NGINX cache. The goal is to reduce the load on the Servers as much as possible.

We use the following NGINX cache control to prevent a thundering herd run on the Server. proxy_cache_lock on; If we get many simultaneous requests before NGINX has the cache, NGINX will block all requests but 1 until the cache is populated. This keeps our Servers safe.

We finally propagate our cache headers all they way back to the CloudFront CDN which is our primary cache point.

What about CORS headers?

We set the following CORS headers on every request.

if ($request_method = 'OPTIONS') {
    add_header 'Access-Control-Allow-Origin' '*';
    add_header 'Access-Control-Allow-Methods' 'GET, OPTIONS';
    #
    # Custom headers and headers various browsers *should* be OK with but aren't
    #
    add_header 'Access-Control-Allow-Headers' 'DNT,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Range';
    #
    # Tell client that this pre-flight info is valid for 20 days
    #
    add_header 'Access-Control-Max-Age' 1728000;
    add_header 'Content-Type' 'text/plain; charset=utf-8';
    add_header 'Content-Length' 0;
    return 204;
}
if ($request_method = 'POST') {
    add_header 'Access-Control-Allow-Origin' '*';
    add_header 'Access-Control-Allow-Methods' 'GET, OPTIONS';
    add_header 'Access-Control-Allow-Headers' 'DNT,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Range';
    add_header 'Access-Control-Expose-Headers' 'Content-Length,Content-Range';
}
if ($request_method = 'GET') {
    add_header 'Access-Control-Allow-Origin' '*';
    add_header 'Access-Control-Allow-Methods' 'GET, OPTIONS';
    add_header 'Access-Control-Allow-Headers' 'DNT,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Range';
    add_header 'Access-Control-Expose-Headers' 'Content-Length,Content-Range';
}

Same as the Proxy we have to keep the fleet of Servers in-sync.

When the service starts we fetch the list of Servers (IP:PORT) and add them to the NGINX configuration. Now we're ready to route traffic.

We then run a cron job to perform this action again, to pick up any new Servers. NGINX is reloaded with zero downtime.

Service

Now it's time to deploy our Origin.

The Origin service has 2 stacks:

ecr - Docker image registry
service - Fargate service

First create the docker ECR registry.

sh ./stack-up.sh ecr

Now we can build, tag, and push the Docker image to the registry.

First update the package.json scripts to include your AWS account id.

To build, tag, and push the Docker image to the registry, run the following command.

yarn run deploy <version>

Now we can deploy the service stack which will deploy our new image to Fargate.

First update the Version here.

Then run:

sh ./stack-up.sh service

Your Origin should now be running in your ECS cluster as a Fargate task.

In conclusion... I think this architecture and implementation works pretty well. It hasn't been battle tested. Here are some things I'd still like to do and some areas for improvement.

1.) Test with more clients & videos.

I tested with VLC, FFMPEG, and my phone. I pushed live video, screen recordings, and VOD's in a loop. It always worked... but it would be good to test with many different clients before going to production.

2.) Load testing.

It would be a good idea to see how this architecture does under load.

RTMP load testing? The limiting factor is the number of Servers. We can only push one stream per server. Not much to test here. Just need to make sure that one Server can handle a large video stream.

HTTP load testing? This could be done pretty easily using something like wkr.

3.) Auto scaling.

This is the big one. If we were to offer a service like Twitch, how would we scale up the number of Servers to meet the growing number of streamers? How would we scale down to save costs and not terminate a users stream.

I think we'd need a custom Controller to perform the scaling and communication between all the services to update their configurations realtime instead of every minute via cron.

That's it. I hope you enjoyed learning about RTMP-to-HLS streaming.

Original published here: https://finbits.io/blog/live-streaming-origin/

Live Streaming Proxy

Randy Findley — Tue, 16 Jun 2020 11:48:18 +0000

If you're just joining us, please take a look at part 1 in this 3-part series.

Part 1 - Server
Part 2 - Proxy
Part 3 - Origin

This blog post is about the Proxy. Remember we have 3 services in this architecture. Proxy -> Server <- Origin

How do we route stateful RTMP traffic to a fleet of RTMP Servers?

We need a single endpoint like rtmp.finbits.io.

A user will push RTMP using a URL like rtmp.finbits.io/stream/<stream key>

RTMP is stateful which makes load balancing and scaling much more challenging.

If this were a stateless application we'd use AWS ALB with an Auto Scaling Group.

Can we use AWS ALB? Nope, ALB doesn't support RTMP.

An Auto Scaling Group might work but scaling down will be a challenge. You wouldn't want to terminate a Server that a user is actively streaming to.

Why not just redirect RTMP directly to the Servers? This would be great but RTMP redirects are fairly new and not all clients support it.

So, how do we route stateful RTMP traffic to a fleet of RTMP Servers?

We can use HAProxy and weighted Route53 DNS routing. We could have n number of Origin (HAProxy) services running, all with public IPs, and then route traffic to all instances via Route53 weighted record sets.

The only gotcha with this service is making sure it picks up any new Servers that are added to handle load.

When the service starts we fetch the list of Servers (IP:PORT) and add them to the HAProxy configuration. Now we're ready to route traffic.

We then run a cron job to perform this action again, to pick up any new Servers. HAProxy is reloaded with zero downtime.

Let's take a look at the HAProxy config.

global
 pidfile /var/run/haproxy.pid
 maxconn <%= servers.length %>

defaults
 log global
 timeout connect 10s
 timeout client 30s
 timeout server 30s

frontend ft_rtpm
 bind *:1935 name rtmp
 mode tcp
 maxconn <%= servers.length %>
 default_backend bk_rtmp

frontend ft_http
 bind *:8000 name http
 mode http
 maxconn 600
 default_backend bk_http

backend bk_http
 mode http
 errorfile 503 /usr/local/etc/haproxy/healthcheck.http

backend bk_rtmp 
 mode tcp
 balance roundrobin
 <% servers.forEach(function(server, index) { %>
 server media<%= index %> <%= server %> check maxconn 1 weight 10
 <% }); %>

As you can see we're using EJS template engine to generate the config.

global

global
 pidfile /var/run/haproxy.pid
 maxconn <%= servers.length %>

We store the pidfile, so we can use it to restart the service when our cron job runs to update the Servers list.

'maxconn' is set to the number of Servers. In our design each Server can only accept 1 connection. FFMPEG draws a lot of CPU and we're using Fargate with lower CPU tasks.

You could use EC2's instead of Fargate with highly performant instance types. Then you could handle more connections per Server.

It might be cool to also use NVIDIA hardware acceleration with FFMPEG, but I didn't get that far. It was getting kinda complicated with Docker.

frontend

frontend ft_rtpm
 bind *:1935 name rtmp
 mode tcp
 maxconn <%= servers.length %>
 default_backend bk_rtmp

Here we declare the RTMP frontend on port 1935. It leverages the RTMP backed below.

backend

backend bk_rtmp 
 mode tcp
 balance roundrobin
 <% servers.forEach(function(server, index) { %>
 server media<%= index %> <%= server %> check maxconn 1 weight 10
 <% }); %>

The backend does the routing magic. In our case it uses a simple roundrobin load balancing algorithm.

http

The http frontend and backend serve a static http response for our Route 53 healthcheck.

Service

Now it's time to deploy our Proxy.

The Proxy service has 2 stacks:

ecr - Docker image registry
service - Fargate service

First create the docker ECR registry.

sh ./stack-up.sh ecr

Now we can build, tag, and push the Docker image to the registry.

First update the package.json scripts to include your AWS account id.

To build, tag, and push the Docker image to the registry, run the following command.

yarn run deploy <version>

Now we can deploy the service stack which will deploy our new image to Fargate.

First update the Version here.

Then run:

sh ./stack-up.sh service

Your Proxy should now be running in your ECS cluster as a Fargate task.

Now on to Part 3, the Origin, to learn how we route HTTP traffic to the appropriate HLS Server.

Original published here: https://finbits.io/blog/live-streaming-proxy/

Live Streaming Server

Randy Findley — Tue, 16 Jun 2020 11:34:58 +0000

I decided to build a live streaming server that accepts RTMP input and outputs Adaptive Bitrate (ABR) HLS.

https://github.com/rgfindl/live-streaming-server

I wanted users to be able to stream anytime using their private stream key. Much like how Twitch, Facebook, and YouTube do it.

I also wanted the live stream recorded and I wanted the user to be able to relay their live stream to other destinations like Twitch, Facebook, and YouTube.

Here is a screenshot of my stream playing in the browser, facebook, twitch, and youtube.

The final architecture is actually 3 services: Proxy -> Server <- Origin

I will cover the Proxy and the Origin in posts 2 and 3 in this blog post series.

Take a look at the architecture:

All 3 services are running as Docker containers on AWS Fargate.

RTMP is sent to the Proxy at rtmp.finbits.io.

HLS is served by the Origin at live.finbits.io.

The Redis cache stores the stream key to Server mapping so the Origin knows which Server to fetch the HLS from. We could have many Servers to meet demand.

S3 is used to store the recordings. The recordings are single bitrate HLS. The largest bitrate from the ABR.live

All 3 services scale independently to meet demand. The Server would scale the most. Transcoding RTMP into ABR HLS is very CPU intensive.

Node Media Server

For the RTMP Server I decided to use a fork of Node Media Server.

https://github.com/rgfindl/Node-Media-Server

Node Media Server accepts RTMP on port 1935. FFMPEG is then used to transcode the RTMP input into HLS. FFMPEG is also used to relay to social media destinations.

Why Node Media Server?

It is actively maintained, it has a lot of github stars, and I like node.js.

I first tried nginx-rtmp-module, because nginx is great. But I couldn't get the social relay working the way I wanted. Also, this project is no longer maintained and is pretty old.

I also looked at ossrs/srs, which seems to be based on nginx-rtmp-module. It didn't seem as flexible, maybe because I'm not a c/c++ developer.

Why fork Node Media Server? What did I change?

I added a few more options to the Node Media Server config to get the HLS working. Specifically the config.trans.tasks object.

const config = {
  ...
  trans: {
    tasks: [
      raw: [...], # FFMPEG command
      ouPaths: [...], # HLS output paths
      cleanup: false, # Don't delete the ouPaths, we'll do it later
    ]
  }
}

I'll talk about each of these in more details below.

FFMPEG

FFMPEG is used to transcode the rtmp input into 3 HLS outputs.

640 x 360
842 x 480
720 x 1280

ffmpeg -hide_banner -y -fflags nobuffer -i rtmp://127.0.0.1:1935/stream/test \
  -vf scale=w=640:h=360:force_original_aspect_ratio=decrease -c:a aac -ar 48000 -c:v libx264 -preset veryfast -profile:v main -crf 20 -sc_threshold 0 -g 48 -keyint_min 48 -hls_time 4 -hls_list_size 6 -hls_flags delete_segments -max_muxing_queue_size 1024 -start_number 100 -b:v 800k -maxrate 856k -bufsize 1200k -b:a 96k -hls_segment_filename media/test/360p/%03d.ts media/test/360p.m3u8 \
  -vf scale=w=842:h=480:force_original_aspect_ratio=decrease -c:a aac -ar 48000 -c:v libx264 -preset veryfast -profile:v main -crf 20 -sc_threshold 0 -g 48 -keyint_min 48 -hls_time 4 -hls_list_size 6 -hls_flags delete_segments -max_muxing_queue_size 1024 -start_number 100 -b:v 1400k -maxrate 1498k -bufsize 2100k -b:a 128k -hls_segment_filename media/test/480p/%03d.ts media/test/480p.m3u8 \
  -vf scale=w=1280:h=720:force_original_aspect_ratio=decrease -c:a aac -ar 48000 -c:v libx264 -preset veryfast -profile:v main -crf 20 -sc_threshold 0 -g 48 -keyint_min 48 -hls_time 4 -hls_list_size 6 -hls_flags delete_segments -max_muxing_queue_size 1024 -start_number 100 -b:v 2800k -maxrate 2996k -bufsize 4200k -b:a 128k -hls_segment_filename media/test/720p/%03d.ts media/test/720p.m3u8

What about the ABR playlist file?

We create that once the first HLS playlist file is created. It looks like this:

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360
360p/index.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=1400000,RESOLUTION=842x480
480p/index.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2800000,RESOLUTION=1280x720
720p/index.m3u8

Server

Our Server does the following things:

Takes RTMP input and converts it to HLS
Creates an ABR HLS playlist
Copies the highest bitrate HLS to S3
Relay's RTMP to social destinations based on query parameters
Exposes a hook for stream key validation
Serves HLS via NGINX reverse proxy with cache headers and CORS

app.js

The app.js does most of the work. Let's take a look at that file in it's entirety.

const NodeMediaServer = require('node-media-server');
const _ = require('lodash');
const { join } = require('path');
const querystring = require('querystring');
const fs = require('./lib/fs');
const hls = require('./lib/hls');
const abr = require('./lib/abr');
const ecs = require('./lib/ecs');
const cache = require('./lib/cache');
const logger = require('./lib/logger');
const utils = require('./lib/utils');

const LOG_TYPE = 4;
logger.setLogType(LOG_TYPE);

// init RTMP server
const init = async () => {
  try {
    // Fetch the container server address (IP:PORT)
    // The IP is from the EC2 server.  The PORT is from the container.
    const SERVER_ADDRESS = process.env.NODE_ENV === 'production' ? await ecs.getServer() : '';

    // Set the Node-Media-Server config.
    const config = {
      logType: LOG_TYPE,
      rtmp: {
        port: 1935,
        chunk_size: 60000,
        gop_cache: true,
        ping: 30,
        ping_timeout: 60
      },
      http: {
        port: 8080,
        mediaroot: process.env.MEDIA_ROOT || 'media',
        allow_origin: '*',
        api: true
      },
      auth: {
        api: false
      },
      relay: {
        ffmpeg: process.env.FFMPEG_PATH || '/usr/local/bin/ffmpeg',
        tasks: [
          {
            app: 'stream',
            mode: 'push',
            edge: 'rtmp://127.0.0.1/hls',
          },
        ],
      },
      trans: {
        ffmpeg: process.env.FFMPEG_PATH || '/usr/local/bin/ffmpeg',
        tasks: [
          {
            app: 'hls',
            hls: true,
            raw: [
              '-vf',
              'scale=w=640:h=360:force_original_aspect_ratio=decrease',
              '-c:a',
              'aac',
              '-ar',
              '48000',
              '-c:v',
              'libx264',
              '-preset',
              'veryfast',
              '-profile:v',
              'main',
              '-crf',
              '20',
              '-sc_threshold',
              '0',
              '-g',
              '48',
              '-keyint_min',
              '48',
              '-hls_time',
              '6',
              '-hls_list_size',
              '10',
              '-hls_flags',
              'delete_segments',
              '-max_muxing_queue_size',
              '1024',
              '-start_number',
              '${timeInMilliseconds}',
              '-b:v',
              '800k',
              '-maxrate',
              '856k',
              '-bufsize',
              '1200k',
              '-b:a',
              '96k',
              '-hls_segment_filename',
              '${mediaroot}/${streamName}/360p/%03d.ts',
              '${mediaroot}/${streamName}/360p/index.m3u8',
              '-vf',
              'scale=w=842:h=480:force_original_aspect_ratio=decrease',
              '-c:a',
              'aac',
              '-ar',
              '48000',
              '-c:v',
              'libx264',
              '-preset',
              'veryfast',
              '-profile:v',
              'main',
              '-crf',
              '20',
              '-sc_threshold',
              '0',
              '-g',
              '48',
              '-keyint_min',
              '48',
              '-hls_time',
              '6',
              '-hls_list_size',
              '10',
              '-hls_flags',
              'delete_segments',
              '-max_muxing_queue_size',
              '1024',
              '-start_number',
              '${timeInMilliseconds}',
              '-b:v',
              '1400k',
              '-maxrate',
              '1498k',
              '-bufsize',
              '2100k',
              '-b:a',
              '128k',
              '-hls_segment_filename',
              '${mediaroot}/${streamName}/480p/%03d.ts',
              '${mediaroot}/${streamName}/480p/index.m3u8',
              '-vf',
              'scale=w=1280:h=720:force_original_aspect_ratio=decrease',
              '-c:a',
              'aac',
              '-ar',
              '48000',
              '-c:v',
              'libx264',
              '-preset',
              'veryfast',
              '-profile:v',
              'main',
              '-crf',
              '20',
              '-sc_threshold',
              '0',
              '-g',
              '48',
              '-keyint_min',
              '48',
              '-hls_time',
              '6',
              '-hls_list_size',
              '10',
              '-hls_flags',
              'delete_segments',
              '-max_muxing_queue_size',
              '1024',
              '-start_number',
              '${timeInMilliseconds}',
              '-b:v',
              '2800k',
              '-maxrate',
              '2996k',
              '-bufsize',
              '4200k',
              '-b:a',
              '128k',
              '-hls_segment_filename',
              '${mediaroot}/${streamName}/720p/%03d.ts',
              '${mediaroot}/${streamName}/720p/index.m3u8'
            ],
            ouPaths: [
              '${mediaroot}/${streamName}/360p',
              '${mediaroot}/${streamName}/480p',
              '${mediaroot}/${streamName}/720p'
            ],
            hlsFlags: '',
            cleanup: false,
          },
        ]
      },
    };

    // Construct the NodeMediaServer
    const nms = new NodeMediaServer(config);

    // Create the maps we'll need to track the current streams.
    this.dynamicSessions = new Map();
    this.streams = new Map();

    // Start the VOD S3 file watcher and sync.
    hls.recordHls(config, this.streams);

    //
    // HLS callbacks
    //
    hls.on('newHlsStream', async (name) => {
      // Create the ABR HLS playlist file.
      await abr.createPlaylist(config.http.mediaroot, name);
      // Send the "stream key" <-> "IP:PORT" mapping to Redis
      // This tells the Origin which Server has the HLS files
      await cache.set(name, SERVER_ADDRESS);
    });

    //
    // RTMP callbacks
    //
    nms.on('preConnect', (id, args) => {
      logger.log('[NodeEvent on preConnect]', `id=${id} args=${JSON.stringify(args)}`);
      // Pre connect authorization
      // let session = nms.getSession(id);
      // session.reject();
    });

    nms.on('postConnect', (id, args) => {
      logger.log('[NodeEvent on postConnect]', `id=${id} args=${JSON.stringify(args)}`);
    });

    nms.on('doneConnect', (id, args) => {
      logger.log('[NodeEvent on doneConnect]', `id=${id} args=${JSON.stringify(args)}`);
    });

    nms.on('prePublish', (id, StreamPath, args) => {
      logger.log('[NodeEvent on prePublish]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
      // Pre publish authorization
      // let session = nms.getSession(id);
      // session.reject();
    });

    nms.on('postPublish', async (id, StreamPath, args) => {
      logger.log('[NodeEvent on postPublish]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
      if (StreamPath.indexOf('/hls/') != -1) {
        // Set the "stream key" <-> "id" mapping for this RTMP/HLS session
        // We use this when creating the DVR HLS playlist name on S3.
        const name = StreamPath.split('/').pop();
        this.streams.set(name, id);
      } else if (StreamPath.indexOf('/stream/') != -1) {
        //
        // Start Relay to youtube, facebook, and/or twitch
        //
        if (args.youtube) {
          const params = utils.getParams(args, 'youtube_');
          const query = _.isEmpty(params) ? '' : `?${querystring.stringify(params)}`;
          const url = `rtmp://a.rtmp.youtube.com/live2/${args.youtube}${query}`;
          const session = nms.nodeRelaySession({
            ffmpeg: config.relay.ffmpeg,
            inPath: `rtmp://127.0.0.1:${config.rtmp.port}${StreamPath}`,
            ouPath: url
          });
          session.id = `youtube-${id}`;
          session.on('end', (id) => {
            this.dynamicSessions.delete(id);
          });
          this.dynamicSessions.set(session.id, session);
          session.run();
        }
        if (args.facebook) {
          const params = utils.getParams(args, 'facebook_');
          const query = _.isEmpty(params) ? '' : `?${querystring.stringify(params)}`;
          const url = `rtmps://live-api-s.facebook.com:443/rtmp/${args.facebook}${query}`;
          session = nms.nodeRelaySession({
            ffmpeg: config.relay.ffmpeg,
            inPath: `rtmp://127.0.0.1:${config.rtmp.port}${StreamPath}`,
            ouPath: url
          });
          session.id = `facebook-${id}`;
          session.on('end', (id) => {
            this.dynamicSessions.delete(id);
          });
          this.dynamicSessions.set(session.id, session);
          session.run();
        }
        if (args.twitch) {
          const params = utils.getParams(args, 'twitch_');
          const query = _.isEmpty(params) ? '' : `?${querystring.stringify(params)}`;
          const url = `rtmp://live-jfk.twitch.tv/app/${args.twitch}${query}`;
          session = nms.nodeRelaySession({
            ffmpeg: config.relay.ffmpeg,
            inPath: `rtmp://127.0.0.1:${config.rtmp.port}${StreamPath}`,
            ouPath: url,
            raw: [
              '-c:v',
              'libx264',
              '-preset',
              'veryfast',
              '-c:a',
              'copy',
              '-b:v',
              '3500k',
              '-maxrate',
              '3750k',
              '-bufsize',
              '4200k',
              '-s',
              '1280x720',
              '-r',
              '30',
              '-f',
              'flv',
              '-max_muxing_queue_size',
              '1024',
            ]
          });
          session.id = `twitch-${id}`;
          session.on('end', (id) => {
            this.dynamicSessions.delete(id);
          });
          this.dynamicSessions.set(session.id, session);
          session.run();
        }
      }
    });

    nms.on('donePublish', async (id, StreamPath, args) => {
      logger.log('[NodeEvent on donePublish]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
      if (StreamPath.indexOf('/hls/') != -1) {
        const name = StreamPath.split('/').pop();
        // Delete the Redis cache key for this stream
        await cache.del(name);
        // Wait a few minutes before deleting the HLS files on this Server
        // for this session
        const timeoutMs = _.isEqual(process.env.NODE_ENV, 'development') ?
          1000 : 
          2 * 60 * 1000;
        await utils.timeout(timeoutMs);
        if (!_.isEqual(await cache.get(name), SERVER_ADDRESS)) {
          // Only clean up if the stream isn't running.  
          // The user could have terminated then started again.
          try {
            // Cleanup directory
            logger.log('[Delete HLS Directory]', `dir=${join(config.http.mediaroot, name)}`);
            this.streams.delete(name);
            fs.rmdirSync(join(config.http.mediaroot, name));
          } catch (err) {
            logger.error(err);
          }
        }
      } else if (StreamPath.indexOf('/stream/') != -1) {
        //
        // Stop the Relay's
        //
        if (args.youtube) {
          let session = this.dynamicSessions.get(`youtube-${id}`);
          if (session) {
            session.end();
            this.dynamicSessions.delete(`youtube-${id}`);
          }
        }
        if (args.facebook) {
          let session = this.dynamicSessions.get(`facebook-${id}`);
          if (session) {
            session.end();
            this.dynamicSessions.delete(`facebook-${id}`);
          }
        }
        if (args.twitch) {
          let session = this.dynamicSessions.get(`twitch-${id}`);
          if (session) {
            session.end();
            this.dynamicSessions.delete(`twitch-${id}`);
          }
        }
      }
    });

    nms.on('prePlay', (id, StreamPath, args) => {
      logger.log('[NodeEvent on prePlay]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
      // Pre play authorization
      // let session = nms.getSession(id);
      // session.reject();
    });

    nms.on('postPlay', (id, StreamPath, args) => {
      logger.log('[NodeEvent on postPlay]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
    });

    nms.on('donePlay', (id, StreamPath, args) => {
      logger.log('[NodeEvent on donePlay]', `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`);
    });

    // Run the NodeMediaServer
    nms.run();
  } catch (err) {
    logger.log('Can\'t start app', err);
    process.exit();
  }
};
init();

Url Structure

When calling the Application your URL would look something like this.

The social query params are optional. When present they Relay to the corresponding social destination.

rtmp://rtmp.finbits.io:1935/stream/testkeyd?twitch=<your twitch key>&youtube=<your youtube key>&facebook=<your facebook key>&facebook_s_bl=<your facebook bl>&facebook_s_sc=<your facebook s_sc>&facebook_s_sw=<your facebook sw>&facebook_s_vt=<your facebook vt>&facebook_a=<your facebook a>

DVR

The highest bitrate HLS is copied to S3.

The bucket path looks like this:

<bucket>/<stream key>/vod-<stream id>.m3u8

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:6
#EXT-X-MEDIA-SEQUENCE:1590665387025
#EXTINF:6.000000,
720p/1590665387025.ts
#EXTINF:6.000000,
720p/1590665387026.ts
#EXT-X-ENDLIST

Stream Key Validation

You can perform stream key validation on either the preConnect or prePublish RTMP events. Here is an example:

nms.on('preConnect', (id, args) => {
  logger.log('[NodeEvent on preConnect]', `id=${id} args=${JSON.stringify(args)}`);
  // Pre connect authorization
  if (isInvalid) {
    let session = nms.getSession(id);
    session.reject();
  }
});

NGINX

We use NGINX as a reverse proxy to serve the static HLS files. It has better performance than express.js.

worker_processes  auto;

error_log /dev/stdout info;


events {
  worker_connections  1024;
}

http {
  include       /etc/nginx/mime.types;
  default_type  application/octet-stream;

  log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                    '$status $body_bytes_sent "$http_referer" '
                    '"$http_user_agent" "$http_x_forwarded_for"';

  access_log /dev/stdout main;

  sendfile        on;

  keepalive_timeout  65;

  gzip on;

  ignore_invalid_headers off;

  upstream node-backend {
    server localhost:8080 max_fails=0;
  }

  server {
    listen 8000;
    server_name localhost;
    sendfile off;

    location ~ live\.m3u8 {
      add_header Cache-Control "max-age=60";
      root /usr/src/app/media;
    }

    location ~ index\.m3u8 {
      add_header Cache-Control "no-cache";
      root /usr/src/app/media;
    }

    location ~ \.ts {
      add_header Cache-Control "max-age=600";
      root /usr/src/app/media;
    }

    location /nginx_status {
      stub_status on;

      access_log off;
      allow 127.0.0.1;
      deny all;
    }

    location / {
      add_header Cache-Control "no-cache";
      proxy_pass http://node-backend/;
    }
  }
}

As you can see we don't cache the HLS playlist files. We do cache the ABR file and the *.ts media files.

Infrastructure

This entire application runs on AWS. Before we can spin up the Proxy, Server, and Origin Fargate services we have to create some shared infrastructure. Here is a list of the shared infrastructure:

assets - S3 Bucket
vpc - VPC for our Fargate services
ecs - ECS cluster for our Fargate services
security - Security Group's for our Fargate services
redis - A Redis cache to store the "stream key" to "IP:PORT" mapping
proxy dns - rtmp.finbits.io DNS

You can use the stack-up.sh script to deploy each of these stacks to your AWS account. You'll have to change the PROFILE="--profile bluefin" to match your credentials file.

Here is an example:

sh ./stack-up.sh vpc

Service

Now that we have the shared infrastructure up. Let's get the Server deployed.

The Server service has 2 stacks:

ecr - Docker image registry
service - Fargate service

First create the docker ECR registry.

sh ./stack-up.sh ecr

Now we can build, tag, and push the Docker image to the registry.

First update the package.json scripts to include your AWS account id.

To build, tag, and push the Docker image to the registry, run the following command.

yarn run deploy <version>

Now we can deploy the service stack which will deploy our new image to Fargate.

First update the Version here.

Then run:

sh ./stack-up.sh service

Your Server should now be running in your ECS cluster as a Fargate task.

But... you can't access it directly. :(

We need a Proxy to route RTMP traffic to our fleet of Servers to publish RTMP.

We also need an Origin to route HTTP traffic to our fleet of Servers.

Take a look at the next blog post in this 3-part series:

Part 1 - Server
Part 2 - Proxy
Part 3 - Origin

Original published here: https://finbits.io/blog/live-streaming-server/

Headless Wordpress CMS on AWS

Randy Findley — Sun, 04 Nov 2018 17:26:56 +0000

Quickly spin up a headless Wordpress CMS on AWS using Docker

Headless CMS is very popular at the moment. But what is a headless CMS and why should I start using one?

A headless CMS is a backend that is decoupled from its frontends. The backend is where the content is created and published. Whereas the frontends are where the content is displayed (web, mobile apps, set-top box, Alexa, etc…).

For example, a traditional CMS is a single website. The same website is used to add the content as it is to display the content. The backend and frontend are coupled.

A headless CMS is only used to created and publish content. That content is then available through an API. The website or mobile apps, for displaying the content, are separate.

But why is decoupled better? Here are some of the reasons:

Faster and more flexible content delivery than traditional CMS
Resiliency in the face of changes on the user interface side (future-proof)
Rapid design iterations
Enhanced security
Fewer publisher and developer dependencies
Simpler deployment

*** Thanks for the list Brightspot

Now that we all emphatically agree that headless CMS is the way to go. Let’s take a look at these awesome CloudFormation templates I created to help you spin up and manage your headless (Wordpress) CMS.

To be clear, this is a Wordpress installation running on AWS using Infrastructure as Code and Docker.

Wait… Why use Wordpress and not one of these cool new headless CMS services like Contently or Cosmic JS? Well, these services are really great but they cost a lot of money, and I usually like to run everything myself, if I can help it. And… Wordpress is really good at managing content.

But how can Wordpress be a headless CMS? Easy, it has an API.

The trick is making Wordpress stateless so that it can autoscale, and run on AWS, with zero-downtime deployments.

Ok, headless CMS is cool, Wordpress is pretty cool, let's continue.

https://github.com/rgfindl/headless-wordpress

Architecture

Our Wordpress instance is running as an Elastic Container Service using Docker.

First, we have the VPC, then an ECS cluster, then an EC2 instance, and finally an ECS service with tasks. Our Wordpress service is exposed to the world via an Elastic Load Balancer. We’re using RDS as our MySQL database.

Our Wordpress service is stateless, which means we can’t rely on the file system to store content, like Wordpress Media or Plugins. Every time an instance of our Wordpress service is spawned it will only have the files that are baked into our Docker image.

Let’s take a look at how we handle Wordpress Media first.

We use a plugin called WP Offload Media. This plugin allows us to store the Media in S3 and use CloudFront as the CDN. Take a look at the diagram below. We also use the same CDN to cache the Wordpress API…

Now how do we handle the Plugins? (We can ignore Templates because this is headless 😃)

Remember when I talked about baking things into our Docker image? That’s it… We have to include the Plugins in our Docker image. Let’s take a look at that Dockerfile and go through it.

FROM wordpress:latest

# Install unzip
RUN apt-get update; \
    apt-get install -y --no-install-recommends unzip

# Install WP plugins
RUN curl -L https://downloads.wordpress.org/plugin/amazon-s3-and-cloudfront.2.0.zip -o /tmp/amazon-s3-and-cloudfront.2.0.zip
RUN unzip /tmp/amazon-s3-and-cloudfront.2.0.zip -d /usr/src/wordpress/wp-content/plugins
RUN rm /tmp/amazon-s3-and-cloudfront.2.0.zip

RUN curl -L https://downloads.wordpress.org/plugin/advanced-custom-fields.5.7.7.zip -o /tmp/advanced-custom-fields.5.7.7.zip
RUN unzip /tmp/advanced-custom-fields.5.7.7.zip -d /usr/src/wordpress/wp-content/plugins
RUN rm /tmp/advanced-custom-fields.5.7.7.zip

RUN curl -L https://downloads.wordpress.org/plugin/custom-post-type-ui.1.5.8.zip -o /tmp/custom-post-type-ui.1.5.8.zip
RUN unzip /tmp/custom-post-type-ui.1.5.8.zip -d /usr/src/wordpress/wp-content/plugins
RUN rm /tmp/custom-post-type-ui.1.5.8.zip

As you can see our Dockerfile is really simple. It extends the latest Wordpress image and then installs 3 Plugins. It downloads, unzips, and copies each plugin to the wordpress/wp-content/ directory. When you launch your Wordpress site for the first time you’ll have to activate these plugins. The activation status is stored in MySQL so you won’t have to do it every time your ECS tasks recycle.

Installation

Alright, let’s get this architecture installed. First a few prerequisites.

Prerequisites

Install the following prerequisites:

AWS Account
EC2 Key Pair
cim - (npm install -g cim)
AWS CLI

Stacks

VPC
ECS
RDS
ECR
Wordpress

VPC

This creates the Amazon Virtual Private Cloud that our ECS cluster and RDS database will run in.

Amazon Virtual Private Cloud (Amazon VPC) lets you provision a logically isolated section of the AWS Cloud where you can launch AWS resources in a virtual network that you define.

cd vpc
cim stack-up

ECS

This creates an Elastic Container Service that our EC2's will run in.

Amazon Elastic Container Service (Amazon ECS) is a highly scalable, high-performance container orchestration service that supports Docker containers and allows you to easily run and scale containerized applications on AWS.

cd vpc
cim stack-up

RDS

This creates a Relational Database Service database cluster that our Wordpress application will use.

Amazon Relational Database Service (Amazon RDS) makes it easy to set up, operate, and scale a relational database in the cloud.

cd rds
export DatabaseUsername="???"; export DatabasePassword="???"; cim stack-up

ECR

This creates an Elastic Container Registry that will hold the docker images of our wordpress service.

Amazon Elastic Container Registry (ECR) is a fully-managed Docker container registry that makes it easy for developers to store, manage, and deploy Docker container images.

cd ecr
cim stack-up

Wordpress

Before we can launch this cloudformation stack. We need to push our service image to ECR.

Push Image

cd wordpress/src

Registry Authentication
- aws ecr get-login --registry-ids <account-id>
- copy/past output to perform docker login, also append /headless-wp to the repository url.
Build Image
- docker build -t headless-wp:<version> .
Push Image
- docker tag headless-wp:<version> <account-id>.dkr.ecr.<region>.amazonaws.com/headless-wp:latest
- docker tag headless-wp:<version> <account-id>.dkr.ecr.<region>.amazonaws.com/headless-wp:<version>
- docker push <account-id>.dkr.ecr.<region>.amazonaws.com/headless-wp

Update Version

Make sure the Version parameter, in _cim.yml, matches the version tag from above. The ECS Task Definition will pull the image from ECR.

Stack up

Once the Version is set you can use cim stack-up to update the stack with the new version.

cd wordpress
cim stack-up

Congratulations, your new Wordpress site is now available.

First run through the Wordpress setup wizard.

Next enable some of the plugins we added.

Add a few blog posts and pages.

Then check out the API. Ex: https://<cdn-url>/wp-json/wp/v2/posts

Environment Variables

Congratulations on getting your Headless Wordpress installed. If you got stuck anywhere along the way please don’t hesitate to reach out to me for help.

One thing I want to explain is the Wordpress environment variables because they really tie everything together. They tell our Wordpress installation about the RDS database, the Media S3 bucket, and CloudFront CDN URL. Let’s take a look. These can be found in the Wordpress stack’s AWS::ECS::TaskDefinition.

Environment:
  - Name: AWS_REGION
    Value: !Ref AWS::Region
  - Name: AWS_ACCOUNT_ID
    Value: !Ref AWS::AccountId
  - Name: WORDPRESS_DB_HOST
    Value:
      Fn::ImportValue:
        !Sub "${RDSStack}-RDSClusterEndpoint"
  - Name: WORDPRESS_DB_USER
    Value:
      Fn::ImportValue:
        !Sub "${RDSStack}-DatabaseUsername"
  - Name: WORDPRESS_DB_PASSWORD
    Value:
      Fn::ImportValue:
        !Sub "${RDSStack}-DatabasePassword"
  - Name: WORDPRESS_CONFIG_EXTRA
    Value: !Sub
      - |
        define( 'AS3CF_AWS_USE_EC2_IAM_ROLE', true );
        define( 'AS3CF_SETTINGS', serialize( array(
          'bucket' => '${MediaBucket}',
          'copy-to-s3' => true,
          'serve-from-s3' => true,
          'domain' => 'cloudfront',
          'cloudfront' => '${DomainName}',
          'enable-object-prefix' => true,
          'object-prefix' => 'wp-content/uploads/',
          'force-https' => ${ForceHttps},
          'remove-local-file' => true
        ) ) );
        define( 'WP_HOME', '${CMSUrl}' );
        define( 'WP_SITEURL', '${CMSUrl}' );
      - {
        MediaBucket: !Ref MediaBucket,
        DomainName: !If [UseCustomDomain, !Ref Domain, !GetAtt CDN.DomainName],
        ForceHttps: !If [UseCustomDomain, 'true', 'false'],
        CMSUrl: !If [UseCustomDomain, { "Fn::ImportValue" : {"Fn::Sub": "${ECSStack}-CustomDomainUrl" } }, {"Fn::Sub": ["http://${Url}", {"Url": { "Fn::ImportValue" : {"Fn::Sub": "${ECSStack}-LoadBalancerUrl" } }}]}]
        }
  - Name: WORDPRESS_DEBUG
    Value: '1'

The WORDPRESS_DB_* variables are straight from the RDS stack. CloudFormation allows you to Export Output variables which can be imported in other stacks.

The WORDPRESS_CONFIG_EXTRA variable is where we configure the WP Offload Media plugin. First we tell it to use our Task Role, AWS::IAM::Role, via the AS3CF_AWS_USE_EC2_IAM_ROLE var. Then we the AS3CF_SETTINGS var to setup the plugin.

Thanks for reading. I hope you enjoyed it!