add batch service to trigger transcoding only

Author: Pengfei Qu

CMakeLists.txt

@@ -14,6 +14,14 @@ if (NOT DEFINED NLIVES)
     set(NLIVES "1")
 endif()
 
+if (NOT DEFINED SCENARIO)
+    set(SCENARIO "cdn")
+endif()
+
+if (SCENARIO STREQUAL "batch")
+    set(NLIVES "0")
+endif()
+
 file(GLOB dirs "deployment" "*")
 list(REMOVE_DUPLICATES dirs)
 foreach(dir ${dirs})
@@ -24,5 +32,6 @@ endforeach()
 
 # legal message
 execute_process(COMMAND printf "\nThis script will build third party components licensed under various open source licenses into your container images. The terms under which those components may be used and distributed can be found with the license document that is provided with those components. Please familiarize yourself with those terms to ensure your distribution of those components complies with the terms of those licenses.\n\n")
-execute_process(COMMAND printf "\n-- Setting: NVODS=${NVODS}, NLIVES=${NLIVES}\n")
+execute_process(COMMAND printf "\n-- Setting: SCENARIO=${SCENARIO}\n")
+execute_process(COMMAND printf "-- Setting: NVODS=${NVODS}, NLIVES=${NLIVES}\n")
 execute_process(COMMAND printf "-- Setting: REGISTRY=${REGISTRY}\n")

batch/.dockerignore

@@ -0,0 +1,3 @@
+CMakeLists.txt
+*.m4
+test/*

batch/CMakeLists.txt

@@ -0,0 +1,2 @@
+set(service "ovc_batch_service")
+include("${CMAKE_SOURCE_DIR}/script/service.cmake")

batch/Dockerfile

@@ -0,0 +1,22 @@
+
+FROM openvisualcloud/xeon-ubuntu1804-media-nginx:20.7
+
+Run DEBIAN_FRONTEND=noninteractive apt-get update && apt-get install -y -q --no-install-recommends python3-setuptools python3-redis python-celery-common python3-tornado python3-kafka python3-kazoo openssh-server && rm -rf /var/lib/apt/lists/*
+
+COPY *.py   /home/
+COPY *.json   /home/
+COPY *.conf /etc/nginx/
+CMD  ["/bin/bash","-c","/home/main.py&/usr/local/sbin/nginx"]
+WORKDIR /home
+
+####
+ARG  UID
+ARG  GID
+## must use ; here to ignore user exist status code
+RUN  [ ${GID} -gt 0 ] && groupadd -f -g ${GID} docker; \
+     [ ${UID} -gt 0 ] && useradd -d /home/docker -g ${GID} -K UID_MAX=${UID} -K UID_MIN=${UID} docker; \
+     touch /var/run/nginx.pid && \
+     mkdir -p /var/log/nginx /var/lib/nginx /var/www/video /var/www/archive && \
+     chown -R ${UID}:${GID} /var/run/nginx.pid /var/www /var/log/nginx /var/lib/nginx
+USER ${UID}
+####

batch/build.sh

@@ -0,0 +1,6 @@
+#!/bin/bash -e
+
+IMAGE="ovc_batch_service"
+DIR=$(dirname $(readlink -f "$0"))
+
+. "${DIR}/../script/build.sh"

batch/main.py

@@ -0,0 +1,43 @@
+#!/usr/bin/python3
+
+from os.path import isfile
+from messaging import Producer
+import time
+from os import listdir
+import json
+
+KAFKA_TOPIC_VODS = "content_provider_sched_vods"
+KAFKA_TOPIC_LIVES = "content_provider_sched_lives"
+DASHLS_ROOT = "/var/www"
+ARCHIVE_ROOT = "/var/www/archive"
+
+config_file="/home/transcoding.json"
+
+streams = [s for s in listdir(ARCHIVE_ROOT) if s.endswith((".mp4", ".avi"))]
+
+info={}
+with open(config_file,"rt") as fd:
+    info=json.load(fd)
+
+print(info,flush=True)
+
+producer = Producer()
+for stream in info[0]["vods"]:
+    # schedule producing the stream
+    if stream["name"] in streams:
+        msg=stream
+        print("start VOD transccoding on {} with {}: ".format(stream["name"],stream["type"]), flush=True)
+        print(msg,flush=True)
+        producer.send(KAFKA_TOPIC_VODS, json.dumps(msg))
+        # wait until file is available, return it
+        start_time = time.time()
+        while time.time() - start_time < 60:
+            if isfile(DASHLS_ROOT+"/"+stream["name"]): break 
+            time.sleep(1)
+
+producer.close()
+
+while True:
+    print("Running...",flush=True)
+    time.sleep(30)
+

batch/messaging.py

@@ -0,0 +1,77 @@
+#!/usr/bin/python3
+
+import socket
+from kafka import KafkaProducer, KafkaConsumer, TopicPartition
+import traceback
+import socket
+import time
+
+KAFKA_HOSTS = ["kafka-service:9092"]
+
+class Producer(object):
+    def __init__(self):
+        super(Producer, self).__init__()
+        self._client_id = socket.gethostname()
+        self._producer = None
+
+    def send(self, topic, message):
+        if not self._producer:
+            try:
+                self._producer = KafkaProducer(bootstrap_servers=KAFKA_HOSTS,
+                                               client_id=self._client_id,
+                                               api_version=(0, 10), acks=0)
+            except:
+                print(traceback.format_exc(), flush=True)
+                self._producer = None
+
+        try:
+            self._producer.send(topic, message.encode('utf-8'))
+        except:
+            print(traceback.format_exc(), flush=True)
+
+    def flush(self):
+        if self._producer:
+            self._producer.flush()
+
+    def close(self):
+        if self._producer:
+            self._producer.close()
+            self._producer=None
+
+class Consumer(object):
+    def __init__(self, group=None):
+        super(Consumer, self).__init__()
+        self._client_id = socket.gethostname()
+        self._group = group
+
+    def messages(self, topic, timeout=None):
+        c = KafkaConsumer(topic, bootstrap_servers=KAFKA_HOSTS, client_id=self._client_id,
+                          group_id=self._group, auto_offset_reset="earliest", api_version=(0, 10))
+
+        for msg in c:
+            yield msg.value.decode('utf-8')
+        c.close()
+
+    def debug(self, topic):
+        c = KafkaConsumer(bootstrap_servers=KAFKA_HOSTS, client_id=self._client_id,
+                          group_id=None, api_version=(0, 10))
+
+        # assign/subscribe topic
+        partitions = c.partitions_for_topic(topic)
+        if not partitions:
+            raise Exception("Topic "+topic+" not exist")
+        c.assign([TopicPartition(topic, p) for p in partitions])
+
+        # seek to beginning if needed
+        c.seek_to_beginning()
+
+        # fetch messages
+        while True:
+            partitions = c.poll(100)
+            if partitions:
+                for p in partitions:
+                    for msg in partitions[p]:
+                        yield msg.value.decode('utf-8')
+            yield ""
+
+        c.close()

batch/nginx.conf

@@ -0,0 +1,46 @@
+
+worker_processes auto;
+worker_rlimit_nofile 8192;
+daemon off;
+error_log /var/www/log/error.log warn;
+
+events {
+    worker_connections  4096;
+}
+
+rtmp {
+    server {
+        listen 1935;
+        chunk_size 4000;
+
+        application stream {
+            live on;
+        }
+
+        application hls {
+            live on;
+            hls on;
+            hls_path /var/www/video/hls;
+            hls_nested on;
+            hls_fragment 3;
+            hls_playlist_length 60;
+            hls_variant _low BANDWIDTH=2048000 RESOLUTION=854x480;
+            hls_variant _mid BANDWIDTH=4096000 RESOLUTION=1280x720;
+            hls_variant _hi  BANDWIDTH=8192000 RESOLUTION=1920x1080;
+        }
+
+        application dash {
+            live on;
+            dash on;
+            dash_path /var/www/video/dash;
+            dash_fragment 4;
+            dash_playlist_length 120;
+            dash_nested on;
+            dash_repetition on;
+            dash_cleanup on;
+            dash_variant _low bandwidth="2048000" width="854" height="480";
+            dash_variant _med bandwidth="4096000" width="1280" height="720";
+            dash_variant _hi bandwidth="8192000" width="1920" height="1080" max;
+        }
+    }
+}

batch/shell.sh

@@ -0,0 +1,7 @@
+#!/bin/bash -e
+
+IMAGE="ovc_batch_service"
+DIR=$(dirname $(readlink -f "$0"))
+OPTIONS=("--volume=${DIR}/../../volume/video/archive:/var/www/archive:ro" "--volume=${DIR}/../../volume/video/dash:/var/www/dash:ro" "--volume=${DIR}/../../volume/video/hls:/var/www/hls:ro")
+
+. "${DIR}/../script/shell.sh"

batch/transcoding.json

@@ -0,0 +1,54 @@
+[{
+    "vods": [{
+        "name": "bbb_sunflower_1080p_30fps_normal.mp4",
+        "type": "hls",
+        "parameters": [
+            [3840, 2160, 14000000, 192000],
+            [2560, 1440, 10000000, 192000],
+            [1920, 1080, 5000000, 192000],
+            [1280, 720, 2800000, 192000],
+            [842, 480, 1400000, 128000],
+            [640, 360, 800000, 128000]
+        ],
+        "codec": "AVC",
+        "loop": 0,
+        "target": "file",
+        "platform": "software"
+    },{
+        "name": "bbb_sunflower_1080p_30fps_normal.mp4",
+        "type": "dash",
+        "parameters": [
+            [3840, 2160, 14000000, 192000],
+            [2560, 1440, 10000000, 192000],
+            [1920, 1080, 5000000, 192000],
+            [1280, 720, 2800000, 192000],
+            [842, 480, 1400000, 128000],
+            [640, 360, 800000, 128000]
+        ],
+        "codec": "AVC",
+        "loop": 0,
+        "target": "file",
+        "platform": "software"
+    }],
+    "lives":  [{
+        "name": "bbb_sunflower_1080p_30fps_normal.mp4",
+        "type": "hls",
+        "parameters": [
+            [1920, 1080, 5000000, 192000]
+        ],
+        "codec": "AVC",
+        "loop": 1,
+        "target": "rtmp",
+        "platform": "software"
+    },{
+        "name": "bbb_sunflower_1080p_30fps_normal.mp4",
+        "type": "dash",
+        "parameters": [
+            [1920, 1080, 5000000, 192000]
+        ],
+        "codec": "AVC",
+        "loop": 1,
+        "target": "rtmp",
+        "platform": "software"
+    }]
+}]

cdn-server/schedule.py

@@ -4,19 +4,29 @@
 from tornado import web, gen
 from messaging import Producer
 import time
+import json
 
-KAFKA_TOPIC = "content_provider_sched"
+KAFKA_TOPIC_VODS = "content_provider_sched_vods"
 DASHLS_ROOT = "/var/www"
 
 class ScheduleHandler(web.RequestHandler):
     @gen.coroutine
     def get(self):
-        stream = self.request.uri.replace("/schedule/", "")
+        stream = self.requeist.uri.replace("/schedule/", "")
 
         # schedule producing the stream
         print("request received to process stream: "+stream, flush=True)
         producer = Producer()
-        producer.send(KAFKA_TOPIC, stream)
+        msg.update({
+            "name":stream.split("/")[1],
+            "type":stream.split("/")[0],
+            "parameters": [ ],
+            "codec": "AVC",
+            "loop": 0,
+            "target": "file",
+            "platform": "software"
+            })
+        producer.send(KAFKA_TOPIC_VODS, msg)
         producer.close()
 
         # wait until file is available, return it

deployment/docker-swarm/build.sh

@@ -2,9 +2,10 @@
 
 DIR=$(dirname $(readlink -f "$0"))
 NVODS="${1:-1}"
-REGISTRY="$3"
+SCENARIO="${3:-cdn}"
+REGISTRY="$4"
 
 rm -rf "$DIR/../../volume/video/cache"
 mkdir -p "$DIR/../../volume/video/cache/hls" "$DIR/../../volume/video/cache/dash"
 
-m4 -DNVODS=${NVODS} -DREGISTRY_PREFIX=${REGISTRY} -I "${DIR}" "${DIR}/docker-compose.yml.m4" > "${DIR}/docker-compose.yml"
+m4 -DNVODS=${NVODS} -DSCENARIO=${SCENARIO} -DREGISTRY_PREFIX=${REGISTRY} -I "${DIR}" "${DIR}/docker-compose.yml.m4" > "${DIR}/docker-compose.yml"

deployment/kubernetes/helm/build.sh

@@ -3,13 +3,14 @@
 DIR=$(dirname $(readlink -f "$0"))
 NVODS="${1:-1}"
 NLIVES="${2:-1}"
-REGISTRY="$3"
+SCENARIO="${3:-cdn}"
+REGISTRY="$4"
 HOSTIP=$(ip route get 8.8.8.8 | awk '/ src /{split(substr($0,index($0," src ")),f);print f[2];exit}')
 
 # make sure helm is functional
 helm version >/dev/null 2>/dev/null || exit 0
 
 echo "Generating helm chart"
 . "${DIR}/../volume-info.sh"
-m4 -DREGISTRY_PREFIX=${REGISTRY} -DNVODS=${NVODS} -DNLIVES=${NLIVES} -DUSERID=$(id -u) -DGROUPID=$(id -g) -DHOSTIP=${HOSTIP} $(env | grep _VOLUME_ | sed 's/^/-D/') -I "${DIR}/cdn-transcode" "$DIR/cdn-transcode/values.yaml.m4" > "$DIR/cdn-transcode/values.yaml"
+m4 -DREGISTRY_PREFIX=${REGISTRY} -DNVODS=${NVODS} -DNLIVES=${NLIVES} -DSCENARIO=${SCENARIO} -DUSERID=$(id -u) -DGROUPID=$(id -g) -DHOSTIP=${HOSTIP} $(env | grep _VOLUME_ | sed 's/^/-D/') -I "${DIR}/cdn-transcode" "$DIR/cdn-transcode/values.yaml.m4" > "$DIR/cdn-transcode/values.yaml"
 

deployment/kubernetes/yaml/build.sh

@@ -3,10 +3,12 @@
 DIR=$(dirname $(readlink -f "$0"))
 NVODS="${1:-1}"
 NLIVES="${2:-1}"
-REGISTRY="$3"
+SCENARIO="${3:-cdn}"
+REGISTRY="$4"
 HOSTIP=$(ip route get 8.8.8.8 | awk '/ src /{split(substr($0,index($0," src ")),f);print f[2];exit}')
 
 . "${DIR}/../volume-info.sh"
+echo "NVODS=${NVODS} NLIVES=${NLIVES} SCENARIO=${SCENARIO}"
 for template in $(find "${DIR}" -maxdepth 1 -name "*.yaml.m4" -print); do
-    m4 -DNVODS=${NVODS} -DNLIVES=${NLIVES} -DHOSTIP=${HOSTIP} -DREGISTRY_PREFIX=${REGISTRY} $(env | grep _VOLUME_ | sed 's/^/-D/') -I "${DIR}" "${template}" > "${template/.m4/}"
+    m4 -DNVODS=${NVODS} -DNLIVES=${NLIVES} -DSCENARIO=${SCENARIO} -DHOSTIP=${HOSTIP} -DREGISTRY_PREFIX=${REGISTRY} $(env | grep _VOLUME_ | sed 's/^/-D/') -I "${DIR}" "${template}" > "${template/.m4/}"
 done

deployment/kubernetes/yaml/cdn-deploy.yaml.m4

@@ -20,10 +20,12 @@ spec:
       enableServiceLinks: false
       containers:
         - name: cdn
-          image: defn(`REGISTRY_PREFIX')ovc_cdn_service:latest
+          image: defn(`REGISTRY_PREFIX')`ovc_'defn(`SCENARIO')_service:latest
           imagePullPolicy: IfNotPresent
           ports:
+ifelse(defn(`SCENARIO'),`cdn',`dnl
             - containerPort: 8443
+')dnl
             - containerPort: 1935
           resources:
             limits:
@@ -51,3 +53,4 @@ spec:
             secret:
                secretName: self-signed-certificate
 PLATFORM_NODE_SELECTOR(`Xeon')dnl
+