add common image

Author: Pengfei Qu

Diff for: common/CMakeLists.txt

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

Diff for: common/Dockerfile

@@ -0,0 +1,19 @@
+
+FROM centos:7.6.1810
+
+RUN  yum install -y -q epel-release && yum install -y -q python36-requests python36-ply python36-psutil && rm -rf /var/cache/yum/*
+
+COPY *.py /home/
+ENV  PYTHONIOENCODING=UTF-8
+
+####
+ARG  USER=docker
+ARG  GROUP=docker
+ARG  UID
+ARG  GID
+## must use ; here to ignore user exist status code
+RUN  [ ${GID} -gt 0 ] && groupadd -f -g ${GID} ${GROUP}; \
+     [ ${UID} -gt 0 ] && useradd -d /home -M -g ${GID} -K UID_MAX=${UID} -K UID_MIN=${UID} ${USER}; \
+     chown -R ${UID}:${GID} /home
+####
+

Diff for: common/abr_hls_dash.py

@@ -0,0 +1,124 @@
+#!/usr/bin/python3
+
+import subprocess
+import json
+
+RENDITIONS_SAMPLE = (
+    # resolution  bitrate(kbps)  audio-rate(kbps)
+    [3840, 2160, 14000000, 192000],
+    [2560, 1440, 10000000, 192000],
+    [1920, 1080, 5000000, 192000],
+    [1280, 720, 2800000, 192000],
+    [842, 480, 1400000, 128000],
+    [640, 360, 800000, 128000]
+)
+
+def to_kps(bitrate):
+    return str(int(bitrate/1000))+"k"
+
+def GetABRCommand(in_file, target, streaming_type, renditions=RENDITIONS_SAMPLE, duration=2,
+                  segment_num=0,loop=0):
+    ffprobe_cmd = ["ffprobe", "-v", "quiet", "-print_format", "json",
+                   "-show_streams", in_file]
+
+    process_id = subprocess.Popen(ffprobe_cmd, stdout=subprocess.PIPE)
+    # the `multiprocessing.Process` process will wait until
+    # the call to the `subprocess.Popen` object is completed
+    process_id.wait()
+    clip_info = json.loads(process_id.stdout.read().decode("utf-8"))
+
+    keyframe_interval = 0
+    frame_height = 0
+    clip_v_duration = 0
+    clip_a_duration = 0
+
+    segment_target_duration = duration       # try to create a new segment every X seconds
+    max_bitrate_ratio = 1.07          # maximum accepted bitrate fluctuations
+    rate_monitor_buffer_ratio = 1.5   # maximum buffer size between bitrate conformance checks
+
+    for item in clip_info["streams"]:
+        if item["codec_type"] == "video":
+            keyframe_interval = int(eval(item["avg_frame_rate"])+0.5)
+            frame_height = item["height"]
+            clip_v_duration = eval(item["duration"])
+        if item["codec_type"] == "audio":
+            clip_a_duration = eval(item["duration"])
+
+    if segment_num != 0:
+        segment_duration = (int)((clip_v_duration+2.0)/segment_num)
+        if segment_duration < segment_target_duration:
+            segment_target_duration = segment_duration
+
+    cmd = []
+    cmd_abr = []
+    if loop:
+        cmd_base = ["ffmpeg", "-hide_banner", "-y", "-stream_loop", "0", "-i", in_file]
+    else:
+        cmd_base = ["ffmpeg", "-hide_banner", "-y", "-i", in_file]
+
+    cmd_static = ["-c:v", "libx264", "-profile:v", "main", "-sc_threshold", "0", "-strict", "-2"]
+    cmd_static += ["-g", str(keyframe_interval), "-keyint_min", str(keyframe_interval)]
+    cmd_dash = ["-use_timeline", "1", "-use_template", "1", "-seg_duration",
+                str(segment_target_duration), "-adaptation_sets", "id=0,streams=v"]
+    cmd_hls = ["-hls_time", str(segment_target_duration), "-hls_list_size", "0"]
+    cmd_fade_in_out = ["-an"]
+
+    master_playlist = "#EXTM3U" + "\n" + "#EXT-X-VERSION:3" +"\n" + "#" + "\n"
+
+    count = 0
+    default_threshold = 4
+
+    for item in renditions:
+        width = item[0]
+        height = item[1]
+        v_bitrate = to_kps(item[2])
+        a_bitrate = to_kps(item[3])
+        maxrate = to_kps(item[2] * max_bitrate_ratio)
+        bufsize = to_kps(item[2] * rate_monitor_buffer_ratio)
+        name = str(height) + "p"
+
+        if frame_height < height:
+            continue
+
+        cmd_1 = []
+        cmd_2 = []
+        cmd_3 = []
+        cmd_4 = []
+
+        if streaming_type == "hls":
+            cmd_1 = ["-vf", "scale=w="+str(width)+":"+"h="+str(height)+":"+"force_original_aspect_ratio=decrease"
+                     +","+ "pad=w="+str(width)+":"+"h="+str(height)+":"+"x=(ow-iw)/2"+":"+"y=(oh-ih)/2"]
+            cmd_2 = ["-b:v", v_bitrate, "-maxrate", maxrate, "-bufsize", bufsize]
+            cmd_3 = ["-f", streaming_type]
+            cmd_4 = ["-hls_segment_filename", target+"/"+name+"_"+"%03d.ts", target+"/"+name+".m3u8"]
+            master_playlist += "#EXT-X-STREAM-INF:BANDWIDTH="+str(item[2])+","+"RESOLUTION="+str(width)+"x"+str(height)+"\n"+name+".m3u8"+"\n"
+            cmd_abr += cmd_static + cmd_1 + cmd_2 + cmd_fade_in_out + cmd_3 + cmd_hls + cmd_4
+
+        if streaming_type == "dash":
+            cmd_1 = ["-map", "0:v", "-b:v"+":"+str(count), v_bitrate, "-s:v"+":"+str(count), str(width)+"x"+str(height),
+                     "-maxrate"+":"+str(count), maxrate, "-bufsize"+":"+str(count), bufsize]
+            cmd_2 = ["-an"]
+            cmd_3 = ["-f", streaming_type]
+            cmd_4 = ["-init_seg_name", name+"-init-stream$RepresentationID$.m4s", "-media_seg_name",
+                     name+"-chunk-stream$RepresentationID$-$Number%05d$.m4s", "-y", target+"/"+name+".mpd"]
+            if clip_a_duration == 0:
+                cmd_1 = ["-map", "0:v", "-b:v"+":"+str(count), v_bitrate, "-s:v"+":"+str(count), str(width)+"x"+str(height),
+                         "-maxrate"+":"+str(count), maxrate, "-bufsize"+":"+str(count), bufsize]
+                cmd_2 = []
+            cmd_abr += cmd_1 + cmd_2
+
+        count += 1
+        if default_threshold < count:
+            break
+
+    if streaming_type == "hls":
+        cmd = cmd_base + cmd_abr
+    elif streaming_type == "dash":
+        cmd = cmd_base + cmd_static + cmd_abr +["-f", "dash"] + cmd_dash + ["-y", target+"/"+"index.mpd"]
+
+    #generate master m3u8 file
+    if streaming_type == "hls":
+        with open(target+"/"+"index.m3u8", "w", encoding='utf-8') as f:
+            f.write(master_playlist)
+
+    return cmd

Diff for: common/build.sh

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

Diff for: common/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()

Diff for: common/shell.sh

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

Diff for: common/zkstate.py

@@ -0,0 +1,48 @@
+#!/usr/bin/python3
+
+from kazoo.client import KazooClient
+from kazoo.exceptions import NoNodeError, NodeExistsError
+from kazoo.protocol.states import KazooState
+import traceback
+import time
+
+ZK_HOSTS = 'zookeeper-service:2181'
+
+class ZKState(object):
+    def __init__(self, path, name=None):
+        super(ZKState, self).__init__()
+        options={"max_tries":-1, "max_delay":5, "ignore_expire":True}
+        self._zk = KazooClient(hosts=ZK_HOSTS, connection_retry=options)
+        try:
+            self._zk.start(timeout=3600)
+        except:
+            print(traceback.format_exc(), flush=True)
+        self._path = path
+        self._name="" if name is None else name+"."
+        self._zk.ensure_path(path)
+
+    def processed(self):
+        return self._zk.exists(self._path+"/"+self._name+"complete")
+
+    def process_start(self):
+        if self.processed():
+            return False
+        try:
+            self._zk.create(self._path+"/"+self._name+"processing", ephemeral=True)
+            return True
+        except NodeExistsError: # another process wins
+            return False
+
+    def process_end(self):
+        try:
+            self._zk.create(self._path+"/"+self._name+"complete")
+        except NodeExistsError:
+            pass
+
+    def process_abort(self):
+        # the ephemeral node will be deleted upon close
+        pass
+
+    def close(self):
+        self._zk.stop()
+        self._zk.close()