Merge branch 'master' into windows_nvidia

biz/ffmpeg.py

@@ -6,50 +6,57 @@ from entity.ffmpeg import FfmpegTask
 import logging
 
 from util import ffmpeg, oss
+from telemetry import get_tracer
 
 logger = logging.getLogger('biz/ffmpeg')
 
 
 def parse_ffmpeg_task(task_info, template_info):
-    tasks = []
-    # 中间片段
-    task_params_str = task_info.get("taskParams", "{}")
-    task_params = json.loads(task_params_str)
-    for part in template_info.get("video_parts"):
-        source = parse_video(part.get('source'), task_params, template_info)
-        if not source:
-            logger.warning("no video found for part: " + str(part))
-            continue
-        only_if = part.get('only_if', '')
-        if only_if:
-            if not check_placeholder_exist(only_if, task_params):
-                logger.info("because only_if exist, placeholder: %s not exist, skip part: %s", only_if, part)
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("parse_ffmpeg_task"):
+        tasks = []
+        # 中间片段
+        task_params_str = task_info.get("taskParams", "{}")
+        task_params = json.loads(task_params_str)
+        for part in template_info.get("video_parts"):
+            source = parse_video(part.get('source'), task_params, template_info)
+            if not source:
+                logger.warning("no video found for part: " + str(part))
                 continue
-        sub_ffmpeg_task = FfmpegTask(source)
-        sub_ffmpeg_task.annexb = True
-        sub_ffmpeg_task.frame_rate = template_info.get("frame_rate", 25)
-        for lut in part.get('filters', []):
-            sub_ffmpeg_task.add_lut(os.path.join(template_info.get("local_path"), lut))
-        for audio in part.get('audios', []):
-            sub_ffmpeg_task.add_audios(os.path.join(template_info.get("local_path"), audio))
-        for overlay in part.get('overlays', []):
-            sub_ffmpeg_task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
-        tasks.append(sub_ffmpeg_task)
-    output_file = "out_" + str(time.time()) + ".mp4"
-    task = FfmpegTask(tasks, output_file=output_file)
-    overall = template_info.get("overall_template")
-    task.center_cut = template_info.get("crop_mode", None)
-    task.frame_rate = template_info.get("frame_rate", 25)
-    if overall.get('source', ''):
-        source = parse_video(overall.get('source'), task_params, template_info)
-        task.add_inputs(source)
-    for lut in overall.get('filters', []):
-        task.add_lut(os.path.join(template_info.get("local_path"), lut))
-    for audio in overall.get('audios', []):
-        task.add_audios(os.path.join(template_info.get("local_path"), audio))
-    for overlay in overall.get('overlays', []):
-        task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
-    return task
+            only_if = part.get('only_if', '')
+            if only_if:
+                if not check_placeholder_exist(only_if, task_params):
+                    logger.info("because only_if exist, placeholder: %s not exist, skip part: %s", only_if, part)
+                    continue
+            sub_ffmpeg_task = FfmpegTask(source)
+            sub_ffmpeg_task.annexb = True
+            sub_ffmpeg_task.frame_rate = template_info.get("frame_rate", 25)
+            for effect in part.get('effects', []):
+                sub_ffmpeg_task.add_effect(effect)
+            for lut in part.get('filters', []):
+                sub_ffmpeg_task.add_lut(os.path.join(template_info.get("local_path"), lut))
+            for audio in part.get('audios', []):
+                sub_ffmpeg_task.add_audios(os.path.join(template_info.get("local_path"), audio))
+            for overlay in part.get('overlays', []):
+                sub_ffmpeg_task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
+            tasks.append(sub_ffmpeg_task)
+        output_file = "out_" + str(time.time()) + ".mp4"
+        task = FfmpegTask(tasks, output_file=output_file)
+        overall = template_info.get("overall_template")
+        task.center_cut = template_info.get("crop_mode", None)
+        task.frame_rate = template_info.get("frame_rate", 25)
+        if overall.get('source', ''):
+            source = parse_video(overall.get('source'), task_params, template_info)
+            task.add_inputs(source)
+        for effect in overall.get('effects', []):
+            task.add_effect(effect)
+        for lut in overall.get('filters', []):
+            task.add_lut(os.path.join(template_info.get("local_path"), lut))
+        for audio in overall.get('audios', []):
+            task.add_audios(os.path.join(template_info.get("local_path"), audio))
+        for overlay in overall.get('overlays', []):
+            task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
+        return task
 
 
 def parse_video(source, task_params, template_info):
@@ -83,13 +90,16 @@ def check_placeholder_exist(placeholder_id, task_params):
         return True
     return False
 
+
 def start_ffmpeg_task(ffmpeg_task):
-   for task in ffmpeg_task.analyze_input_render_tasks():
-       result = start_ffmpeg_task(task)
-       if not result:
-           return False
-   ffmpeg_task.correct_task_type()
-   return ffmpeg.start_render(ffmpeg_task)
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_ffmpeg_task"):
+        for task in ffmpeg_task.analyze_input_render_tasks():
+            result = start_ffmpeg_task(task)
+            if not result:
+                return False
+        ffmpeg_task.correct_task_type()
+        return ffmpeg.start_render(ffmpeg_task)
 
 
 def clear_task_tmp_file(ffmpeg_task):

biz/task.py

@@ -1,24 +1,27 @@
+from biz.ffmpeg import parse_ffmpeg_task, start_ffmpeg_task, clear_task_tmp_file, probe_video_info
+from telemetry import get_tracer
 from template import get_template_def
 from util import api
 
 
 def start_task(task_info):
-    from biz.ffmpeg import parse_ffmpeg_task, start_ffmpeg_task, clear_task_tmp_file, probe_video_info
-    task_info = api.normalize_task(task_info)
-    template_info = get_template_def(task_info.get("templateId"))
-    api.report_task_start(task_info)
-    ffmpeg_task = parse_ffmpeg_task(task_info, template_info)
-    result = start_ffmpeg_task(ffmpeg_task)
-    if not result:
-        return api.report_task_failed(task_info)
-    oss_result = api.upload_task_file(task_info, ffmpeg_task)
-    if not oss_result:
-        return api.report_task_failed(task_info)
-    # 获取视频长度宽度和时长
-    width, height, duration = probe_video_info(ffmpeg_task)
-    clear_task_tmp_file(ffmpeg_task)
-    api.report_task_success(task_info, videoInfo={
-        "width": width,
-        "height": height,
-        "duration": duration
-    })
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_task"):
+        task_info = api.normalize_task(task_info)
+        template_info = get_template_def(task_info.get("templateId"))
+        api.report_task_start(task_info)
+        ffmpeg_task = parse_ffmpeg_task(task_info, template_info)
+        result = start_ffmpeg_task(ffmpeg_task)
+        if not result:
+            return api.report_task_failed(task_info)
+        oss_result = api.upload_task_file(task_info, ffmpeg_task)
+        if not oss_result:
+            return api.report_task_failed(task_info)
+        # 获取视频长度宽度和时长
+        width, height, duration = probe_video_info(ffmpeg_task)
+        clear_task_tmp_file(ffmpeg_task)
+        api.report_task_success(task_info, videoInfo={
+            "width": width,
+            "height": height,
+            "duration": duration
+        })

entity/ffmpeg.py

@@ -1,5 +1,6 @@
 import time
 import uuid
+from typing import Any
 
 ENCODER_ARGS = ("-c:v", "h264_nvenc",)
 PROFILE_LEVEL_ARGS = ("-profile:v", "high", "-level:v", "4")
@@ -7,6 +8,8 @@ PROFILE_LEVEL_ARGS = ("-profile:v", "high", "-level:v", "4")
 
 class FfmpegTask(object):
 
+    effects: list[str]
+
     def __init__(self, input_file, task_type='copy', output_file=''):
         self.annexb = False
         if type(input_file) is str:
@@ -29,6 +32,7 @@ class FfmpegTask(object):
         self.luts = []
         self.audios = []
         self.overlays = []
+        self.effects = []
 
     def __repr__(self):
         _str = f'FfmpegTask(input_file={self.input_file}, task_type={self.task_type}'
@@ -40,6 +44,8 @@ class FfmpegTask(object):
             _str += f', overlays={self.overlays}'
         if self.annexb:
             _str += f', annexb={self.annexb}'
+        if self.effects:
+            _str += f', effects={self.effects}'
         if self.mute:
             _str += f', mute={self.mute}'
         return _str + ')'
@@ -83,6 +89,10 @@ class FfmpegTask(object):
         self.luts.extend(luts)
         self.correct_task_type()
 
+    def add_effect(self, *effects):
+        self.effects.extend(effects)
+        self.correct_task_type()
+
     def get_output_file(self):
         if self.task_type == 'copy':
             return self.input_file[0]
@@ -105,6 +115,8 @@ class FfmpegTask(object):
             return False
         if len(self.subtitles) > 0:
             return False
+        if len(self.effects) > 0:
+            return False
         if self.speed != 1:
             return False
         if self.zoom_cut is not None:
@@ -120,6 +132,8 @@ class FfmpegTask(object):
             return False
         if len(self.subtitles) > 0:
             return False
+        if len(self.effects) > 0:
+            return False
         if self.speed != 1:
             return False
         if len(self.audios) > 1:
@@ -141,7 +155,7 @@ class FfmpegTask(object):
         if self.task_type == 'encode':
             input_args = []
             filter_args = []
-            output_args = ["-profile", "high", "-level", "4","-shortest", *ENCODER_ARGS]
+            output_args = [*PROFILE_LEVEL_ARGS,"-shortest", *ENCODER_ARGS]
             if self.annexb:
                 output_args.append("-bsf:v")
                 output_args.append("h264_mp4toannexb")
@@ -162,6 +176,29 @@ class FfmpegTask(object):
                 _f_x = pos_json.get('ltX', 0)
                 _x = f'{float(_f_x/_v_w) :.5f}*iw'
                 filter_args.append(f"[{video_output_str}]crop=x={_x}:y=0:w=ih*ih/iw:h=ih[{video_output_str}]")
+            for effect in self.effects:
+                if effect.startswith("cameraShot:"):
+                    param = effect.split(":", 2)[1]
+                    if param == '':
+                        param = "3,1"
+                    _split = param.split(",")
+                    start = 3
+                    duration = 1
+                    if len(_split) >= 2:
+                        start = int(_split[0])
+                        duration = int(_split[1])
+                    elif len(_split) == 1:
+                        start = int(_split[0])
+                    _start_out_str = "[eff_s]"
+                    _end_out_str = "[eff_e]"
+                    filter_args.append(f"{video_output_str}fps=fps={self.frame_rate},split{_start_out_str}{_end_out_str}")
+                    filter_args.append(f"{_start_out_str}trim=start={0}:end={start+duration},freezeframes=first={start*self.frame_rate}:replace={start*self.frame_rate}{_start_out_str}")
+                    filter_args.append(f"{_end_out_str}trim=start={start}{_end_out_str}")
+                    video_output_str = "[v_eff]"
+                    filter_args.append(f"{_start_out_str}{_end_out_str}concat=n=2:v=1:a=0{video_output_str}")
+                elif effect.startswith("zoom:"):
+                    ...
+                ...
             for lut in self.luts:
                 filter_args.append(f"[{video_output_str}]lut3d=file={lut}[{video_output_str}]")
             for overlay in self.overlays:

index.py

@@ -2,6 +2,7 @@ from time import sleep
 
 import biz.task
 import config
+from telemetry import init_opentelemetry
 from template import load_local_template
 from util import api
 
@@ -12,6 +13,7 @@ load_local_template()
 import logging
 
 LOGGER = logging.getLogger(__name__)
+init_opentelemetry()
 
 while True:
     # print(get_sys_info())

telemetry/__init__.py

@@ -0,0 +1,30 @@
+import os
+
+from constant import SOFTWARE_VERSION
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as OTLPSpanHttpExporter
+from opentelemetry.sdk.resources import DEPLOYMENT_ENVIRONMENT, HOST_NAME, Resource, SERVICE_NAME, SERVICE_VERSION
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+
+def get_tracer(name):
+    return trace.get_tracer(name)
+
+# 初始化 OpenTelemetry
+def init_opentelemetry():
+    # 设置服务名、主机名
+    resource = Resource(attributes={
+        SERVICE_NAME: "RENDER_WORKER",
+        SERVICE_VERSION: SOFTWARE_VERSION,
+        DEPLOYMENT_ENVIRONMENT: "Python",
+        HOST_NAME: os.getenv("ACCESS_KEY"),
+    })
+
+    # 使用HTTP协议上报
+    span_processor = BatchSpanProcessor(OTLPSpanHttpExporter(
+        endpoint="http://tracing-analysis-dc-sh.aliyuncs.com/adapt_e7qojqi4e0@aa79b4d367fb6b7_e7qojqi4e0@53df7ad2afe8301/api/otlp/traces",
+    ))
+
+    trace_provider = TracerProvider(resource=resource, active_span_processor=span_processor)
+    trace.set_tracer_provider(trace_provider)

util/api.py

@@ -1,9 +1,11 @@
+import json
 import logging
 import os
 
 import requests
 
 import util.system
+from telemetry import get_tracer
 
 session = requests.Session()
 logger = logging.getLogger(__name__)
@@ -19,33 +21,41 @@ def sync_center():
     通过接口获取任务
     :return: 任务列表
     """
-    try:
-        from template import TEMPLATES, download_template
-
-        response = session.post(os.getenv('API_ENDPOINT') + "/sync", json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-            'clientStatus': util.system.get_sys_info(),
-            'templateList': [{'id': t.get('id', ''), 'updateTime': t.get('updateTime', '')} for t in TEMPLATES.values()]
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return []
-    data = response.json()
-    logger.debug("获取任务结果：【%s】", data)
-    if data.get('code', 0) == 200:
-        templates = data.get('data', {}).get('templates', [])
-        tasks = data.get('data', {}).get('tasks', [])
-    else:
-        tasks = []
-        templates = []
-        logger.warning("获取任务失败")
-    for template in templates:
-        template_id = template.get('id', '')
-        if template_id:
-            logger.info("更新模板：【%s】", template_id)
-            download_template(template_id)
-    return tasks
+    from template import TEMPLATES, download_template
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("sync_center"):
+        with get_tracer("api").start_as_current_span("sync_center.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url", os.getenv('API_ENDPOINT') + "/sync")
+                response = session.post(os.getenv('API_ENDPOINT') + "/sync", json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                    'clientStatus': util.system.get_sys_info(),
+                    'templateList': [{'id': t.get('id', ''), 'updateTime': t.get('updateTime', '')} for t in
+                                     TEMPLATES.values()]
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                response.raise_for_status()
+                req_span.set_attribute("api.response", response.text)
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return []
+        data = response.json()
+        logger.debug("获取任务结果：【%s】", data)
+        if data.get('code', 0) == 200:
+            templates = data.get('data', {}).get('templates', [])
+            tasks = data.get('data', {}).get('tasks', [])
+        else:
+            tasks = []
+            templates = []
+            logger.warning("获取任务失败")
+        for template in templates:
+            template_id = template.get('id', '')
+            if template_id:
+                logger.info("更新模板：【%s】", template_id)
+                download_template(template_id)
+        return tasks
 
 
 def get_template_info(template_id):
@@ -56,123 +66,177 @@ def get_template_info(template_id):
     :type template_id: str
     :return: 模板信息
     """
-    try:
-        response = session.post('{0}/template/{1}'.format(os.getenv('API_ENDPOINT'), template_id), json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return None
-    data = response.json()
-    remote_template_info = data.get('data', {})
-    logger.debug("获取模板信息结果：【%s】", remote_template_info)
-    template = {
-        'id': template_id,
-        'updateTime': remote_template_info.get('updateTime', template_id),
-        'scenic_name': remote_template_info.get('scenicName', '景区'),
-        'name': remote_template_info.get('name', '模版'),
-        'video_size': '1920x1080',
-        'frame_rate': 25,
-        'overall_duration': 30,
-        'video_parts': [
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("get_template_info"):
+        with get_tracer("api").start_as_current_span("get_template_info.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url", '{0}/template/{1}'.format(os.getenv('API_ENDPOINT'), template_id))
+                response = session.post('{0}/template/{1}'.format(os.getenv('API_ENDPOINT'), template_id), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                response.raise_for_status()
+                req_span.set_attribute("api.response", response.text)
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
+        data = response.json()
+        logger.debug("获取模板信息结果：【%s】", data)
+        remote_template_info = data.get('data', {})
+        template = {
+            'id': template_id,
+            'updateTime': remote_template_info.get('updateTime', template_id),
+            'scenic_name': remote_template_info.get('scenicName', '景区'),
+            'name': remote_template_info.get('name', '模版'),
+            'video_size': '1920x1080',
+            'frame_rate': 25,
+            'overall_duration': 30,
+            'video_parts': [
 
-        ]
-    }
+            ]
+        }
 
-    def _template_normalizer(template_info):
-        _template = {}
-        _placeholder_type = template_info.get('isPlaceholder', -1)
-        if _placeholder_type == 0:
-            # 固定视频
-            _template['source'] = template_info.get('sourceUrl', '')
-        elif _placeholder_type == 1:
-            # 占位符
-            _template['source'] = "PLACEHOLDER_" + template_info.get('sourceUrl', '')
-            _template['mute'] = template_info.get('mute', True)
-            _template['crop_mode'] = template_info.get('cropEnable', None)
-        else:
-            _template['source'] = None
-        _overlays = template_info.get('overlays', '')
-        if _overlays:
-            _template['overlays'] = _overlays.split(",")
-        _audios = template_info.get('audios', '')
-        if _audios:
-            _template['audios'] = _audios.split(",")
-        _luts = template_info.get('luts', '')
-        if _luts:
-            _template['luts'] = _luts.split(",")
-        _only_if = template_info.get('onlyIf', '')
-        if _only_if:
-            _template['only_if'] = _only_if
-        return _template
+        def _template_normalizer(template_info):
+            _template = {}
+            _placeholder_type = template_info.get('isPlaceholder', -1)
+            if _placeholder_type == 0:
+                # 固定视频
+                _template['source'] = template_info.get('sourceUrl', '')
+            elif _placeholder_type == 1:
+                # 占位符
+                _template['source'] = "PLACEHOLDER_" + template_info.get('sourceUrl', '')
+                _template['mute'] = template_info.get('mute', True)
+                _template['crop_mode'] = template_info.get('cropEnable', None)
+            else:
+                _template['source'] = None
+            _overlays = template_info.get('overlays', '')
+            if _overlays:
+                _template['overlays'] = _overlays.split(",")
+            _audios = template_info.get('audios', '')
+            if _audios:
+                _template['audios'] = _audios.split(",")
+            _luts = template_info.get('luts', '')
+            if _luts:
+                _template['luts'] = _luts.split(",")
+            _only_if = template_info.get('onlyIf', '')
+            if _only_if:
+                _template['only_if'] = _only_if
+            _effects = template_info.get('effects', '')
+            if _effects:
+                _template['effects'] = _effects.split("|")
+            return _template
 
-    # outer template definition
-    overall_template = _template_normalizer(remote_template_info)
-    template['overall_template'] = overall_template
-    # inter template definition
-    inter_template_list = remote_template_info.get('children', [])
-    for children_template in inter_template_list:
-        parts = _template_normalizer(children_template)
-        template['video_parts'].append(parts)
-    template['local_path'] = os.path.join(os.getenv('TEMPLATE_DIR'), str(template_id))
-    return template
+        # outer template definition
+        overall_template = _template_normalizer(remote_template_info)
+        template['overall_template'] = overall_template
+        # inter template definition
+        inter_template_list = remote_template_info.get('children', [])
+        for children_template in inter_template_list:
+            parts = _template_normalizer(children_template)
+            template['video_parts'].append(parts)
+        template['local_path'] = os.path.join(os.getenv('TEMPLATE_DIR'), str(template_id))
+        with get_tracer("api").start_as_current_span("get_template_info.template") as res_span:
+            res_span.set_attribute("normalized.response", json.dumps(template))
+        return template
 
 
 def report_task_success(task_info, **kwargs):
-    try:
-        response = session.post('{0}/{1}/success'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-            **kwargs
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return None
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_success"):
+        with get_tracer("api").start_as_current_span("report_task_success.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/success'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/success'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                    **kwargs
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                response.raise_for_status()
+                req_span.set_attribute("api.response", response.text)
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
 
 
 def report_task_start(task_info):
-    try:
-        response = session.post('{0}/{1}/start'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return None
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_start"):
+        with get_tracer("api").start_as_current_span("report_task_start.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/start'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/start'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                response.raise_for_status()
+                req_span.set_attribute("api.response", response.text)
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
 
 
 def report_task_failed(task_info, reason=''):
-    try:
-        response = session.post('{0}/{1}/fail'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-            'reason': reason
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return None
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_failed"):
+        with tracer.start_as_current_span("report_task_failed.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/fail'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/fail'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                    'reason': reason
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                response.raise_for_status()
+                req_span.set_attribute("api.response", response.text)
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
 
 
 def upload_task_file(task_info, ffmpeg_task):
-    logger.info("开始上传文件: %s", task_info.get("id"))
-    try:
-        response = session.post('{0}/{1}/uploadUrl'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
-            'accessKey': os.getenv('ACCESS_KEY'),
-        }, timeout=10)
-        response.raise_for_status()
-    except requests.RequestException as e:
-        logger.error("请求失败！", e)
-        return False
-    data = response.json()
-    url = data.get('data', "")
-    logger.info("开始上传文件: %s 至 %s", task_info.get("id"), url)
-    try:
-        with open(ffmpeg_task.get_output_file(), 'rb') as f:
-            requests.put(url, data=f, headers={"Content-Type": "video/mp4"})
-    except requests.RequestException as e:
-        logger.error("上传失败！", e)
-        return False
-    finally:
-        logger.info("上传文件结束: %s", task_info.get("id"))
-    return True
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("upload_task_file") as span:
+        logger.info("开始上传文件: %s", task_info.get("id"))
+        span.set_attribute("file.id", task_info.get("id"))
+        try:
+            with tracer.start_as_current_span("upload_task_file.request_upload_url") as req_span:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/uploadUrl'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/uploadUrl'.format(os.getenv('API_ENDPOINT'), task_info.get("id")),
+                                        json={
+                                            'accessKey': os.getenv('ACCESS_KEY'),
+                                        }, timeout=10)
+                response.raise_for_status()
+                req_span.set_attribute("http.status_code", response.status_code)
+        except requests.RequestException as e:
+            span.set_attribute("api.error", str(e))
+            logger.error("请求失败！", e)
+            return False
+        data = response.json()
+        url = data.get('data', "")
+        logger.info("开始上传文件: %s 至 %s", task_info.get("id"), url)
+        try:
+            with tracer.start_as_current_span("upload_task_file.start_upload_file") as upload_span:
+                upload_span.set_attribute("http.method", "PUT")
+                upload_span.set_attribute("http.url", url)
+                with open(ffmpeg_task.get_output_file(), 'rb') as f:
+                    requests.put(url, data=f, headers={"Content-Type": "video/mp4"})
+        except requests.RequestException as e:
+            span.set_attribute("api.error", str(e))
+            logger.error("上传失败！", e)
+            return False
+        finally:
+            logger.info("上传文件结束: %s", task_info.get("id"))
+        return True

util/ffmpeg.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 import subprocess
@@ -5,60 +6,81 @@ from datetime import datetime
 from typing import Optional, IO
 
 from entity.ffmpeg import FfmpegTask, ENCODER_ARGS, PROFILE_LEVEL_ARGS
+from telemetry import get_tracer
 
 logger = logging.getLogger(__name__)
 
 def to_annexb(file):
-    if not os.path.exists(file):
-        return file
-    logger.info("ToAnnexb: %s", file)
-    ffmpeg_process = subprocess.run(["ffmpeg", "-y", "-hide_banner", "-i", file, "-c", "copy", "-bsf:v", "h264_mp4toannexb",
-                    "-f", "mpegts", file+".ts"])
-    logger.info("ToAnnexb: %s, returned: %s", file, ffmpeg_process.returncode)
-    if ffmpeg_process.returncode == 0:
-        os.remove(file)
-        return file+".ts"
-    else:
-        return file
+    with get_tracer("ffmpeg").start_as_current_span("to_annexb") as span:
+        span.set_attribute("file.path", file)
+        if not os.path.exists(file):
+            return file
+        logger.info("ToAnnexb: %s", file)
+        ffmpeg_process = subprocess.run(["ffmpeg", "-y", "-hide_banner", "-i", file, "-c", "copy", "-bsf:v", "h264_mp4toannexb",
+                        "-f", "mpegts", file+".ts"])
+        span.set_attribute("ffmpeg.args", json.dumps(ffmpeg_process.args))
+        logger.info("ToAnnexb: %s, returned: %s", file, ffmpeg_process.returncode)
+        span.set_attribute("ffmpeg.code", ffmpeg_process.returncode)
+        if ffmpeg_process.returncode == 0:
+            span.set_attribute("file.size", os.path.getsize(file+".ts"))
+            os.remove(file)
+            return file+".ts"
+        else:
+            return file
 
 def re_encode_and_annexb(file):
-    if not os.path.exists(file):
-        return file
-    logger.info("ReEncodeAndAnnexb: %s", file)
-    ffmpeg_process = subprocess.run(["ffmpeg", "-y", "-hide_banner", "-i", file, *PROFILE_LEVEL_ARGS, *ENCODER_ARGS, "-bsf:v", "h264_mp4toannexb",
-                    "-f", "mpegts", file +".ts"])
-    logger.info("ReEncodeAndAnnexb: %s, returned: %s", file, ffmpeg_process.returncode)
-    if ffmpeg_process.returncode == 0:
-        os.remove(file)
-        return file+".ts"
-    else:
-        return file
+    with get_tracer("ffmpeg").start_as_current_span("re_encode_and_annexb") as span:
+        span.set_attribute("file.path", file)
+        if not os.path.exists(file):
+            return file
+        logger.info("ReEncodeAndAnnexb: %s", file)
+        ffmpeg_process = subprocess.run(["ffmpeg", "-y", "-hide_banner", "-i", file, *PROFILE_LEVEL_ARGS, *ENCODER_ARGS, "-bsf:v", "h264_mp4toannexb",
+                        "-f", "mpegts", file +".ts"])
+        span.set_attribute("ffmpeg.args", json.dumps(ffmpeg_process.args))
+        logger.info("ReEncodeAndAnnexb: %s, returned: %s", file, ffmpeg_process.returncode)
+        span.set_attribute("ffmpeg.code", ffmpeg_process.returncode)
+        if ffmpeg_process.returncode == 0:
+            span.set_attribute("file.size", os.path.getsize(file+".ts"))
+            os.remove(file)
+            return file+".ts"
+        else:
+            return file
 
 def start_render(ffmpeg_task: FfmpegTask):
-    logger.info(ffmpeg_task)
-    if not ffmpeg_task.need_run():
-        ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
-        return True
-    ffmpeg_args = ffmpeg_task.get_ffmpeg_args()
-    logger.info(ffmpeg_args)
-    if len(ffmpeg_args) == 0:
-        ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
-        return True
-    ffmpeg_process = subprocess.run(["ffmpeg", "-progress", "-", "-loglevel", "error", *ffmpeg_args], **subprocess_args(True))
-    logger.info("FINISH TASK, OUTPUT IS %s", handle_ffmpeg_output(ffmpeg_process.stdout))
-    code = ffmpeg_process.returncode
-    if code != 0:
-        logger.error("FFMPEG ERROR: %s", ffmpeg_process.stderr)
-        return False
-    try:
-        out_file_stat = os.stat(ffmpeg_task.output_file)
-        if out_file_stat.st_size < 4096:
-            logger.error("FFMPEG ERROR: OUTPUT FILE IS TOO SMALL")
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_render") as span:
+        span.set_attribute("ffmpeg.task", str(ffmpeg_task))
+        logger.info(ffmpeg_task)
+        if not ffmpeg_task.need_run():
+            ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
+            return True
+        ffmpeg_args = ffmpeg_task.get_ffmpeg_args()
+        logger.info(ffmpeg_args)
+        if len(ffmpeg_args) == 0:
+            ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
+            return True
+        ffmpeg_process = subprocess.run(["ffmpeg", "-progress", "-", "-loglevel", "error", *ffmpeg_args], **subprocess_args(True))
+        span.set_attribute("ffmpeg.args", json.dumps(ffmpeg_process.args))
+        ffmpeg_final_out = handle_ffmpeg_output(ffmpeg_process.stdout)
+        span.set_attribute("ffmpeg.out", ffmpeg_final_out)
+        logger.info("FINISH TASK, OUTPUT IS %s", ffmpeg_final_out)
+        code = ffmpeg_process.returncode
+        span.set_attribute("ffmpeg.code", code)
+        if code != 0:
+            logger.error("FFMPEG ERROR: %s", ffmpeg_process.stderr)
             return False
-    except OSError:
-        logger.error("FFMPEG ERROR: OUTPUT FILE NOT FOUND")
-        return False
-    return True
+        span.set_attribute("ffmpeg.out_file", ffmpeg_task.output_file)
+        try:
+            file_size = os.path.getsize(ffmpeg_task.output_file)
+            span.set_attribute("file.size", file_size)
+            if file_size < 4096:
+                logger.error("FFMPEG ERROR: OUTPUT FILE IS TOO SMALL")
+                return False
+        except OSError:
+            span.set_attribute("file.size", 0)
+            logger.error("FFMPEG ERROR: OUTPUT FILE NOT FOUND")
+            return False
+        return True
 
 def handle_ffmpeg_output(stdout: Optional[bytes]) -> str:
     out_time = "0:0:0.0"
@@ -79,27 +101,31 @@ def handle_ffmpeg_output(stdout: Optional[bytes]) -> str:
     print("[ ]Speed:", out_time, "@", speed)
     return out_time+"@"+speed
 
-
 def duration_str_to_float(duration_str: str) -> float:
     _duration = datetime.strptime(duration_str, "%H:%M:%S.%f") - datetime(1900, 1, 1)
     return _duration.total_seconds()
 
 
 def probe_video_info(video_file):
-    # 获取宽度和高度
-    result = subprocess.run(
-        ["ffprobe", '-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height:format=duration', '-of',
-         'csv=s=x:p=0', video_file],
-        stderr=subprocess.STDOUT,
-        **subprocess_args(True)
-    )
-    if result.returncode != 0:
-        return 0, 0, 0
-    all_result = result.stdout.decode('utf-8').strip()
-    wh, duration = all_result.split('\n')
-    width, height = wh.strip().split('x')
-
-    return int(width), int(height), float(duration)
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("probe_video_info") as span:
+        span.set_attribute("video.file", video_file)
+        # 获取宽度和高度
+        result = subprocess.run(
+            ["ffprobe", '-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height:format=duration', '-of',
+             'csv=s=x:p=0', video_file],
+            stderr=subprocess.STDOUT,
+            **subprocess_args(True)
+        )
+        span.set_attribute("ffprobe.args", json.dumps(result.args))
+        span.set_attribute("ffprobe.code", result.returncode)
+        if result.returncode != 0:
+            return 0, 0, 0
+        all_result = result.stdout.decode('utf-8').strip()
+        span.set_attribute("ffprobe.out", all_result)
+        wh, duration = all_result.split('\n')
+        width, height = wh.strip().split('x')
+        return int(width), int(height), float(duration)
 
 
 # Create a set of arguments which make a ``subprocess.Popen`` (and

util/oss.py

@@ -3,6 +3,8 @@ import os
 
 import requests
 
+from telemetry import get_tracer
+
 logger = logging.getLogger(__name__)
 
 
@@ -13,20 +15,29 @@ def upload_to_oss(url, file_path):
     :param str file_path: 文件路径
     :return bool: 是否成功
     """
-    max_retries = 5
-    retries = 0
-    while retries < max_retries:
-        try:
-            with open(file_path, 'rb') as f:
-                response = requests.put(url, data=f, timeout=60)  # 设置超时时间为1分钟
-                if response.status_code == 200:
-                    return True
-        except requests.exceptions.Timeout:
-            retries += 1
-            logger.warning(f"Upload timed out. Retrying {retries}/{max_retries}...")
-        except Exception as e:
-            logger.warning(f"Upload failed. Retrying {retries}/{max_retries}...")
-            retries += 1
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("upload_to_oss") as span:
+        span.set_attribute("file.url", url)
+        span.set_attribute("file.path", file_path)
+        max_retries = 5
+        retries = 0
+        while retries < max_retries:
+            try:
+                with tracer.start_as_current_span("upload_to_oss.request") as req_span:
+                    req_span.set_attribute("http.method", "PUT")
+                    req_span.set_attribute("http.url", url)
+                    with open(file_path, 'rb') as f:
+                        response = requests.put(url, data=f, timeout=60)  # 设置超时时间为1分钟
+                        req_span.set_attribute("http.status_code", response.status_code)
+                        if response.status_code == 200:
+                            return True
+            except requests.exceptions.Timeout:
+                retries += 1
+                logger.warning(f"Upload timed out. Retrying {retries}/{max_retries}...")
+            except Exception as e:
+                retries += 1
+                span.set_attribute("oss.error", str(e))
+                logger.warning(f"Upload failed. Retrying {retries}/{max_retries}...")
     return False
 
 
@@ -37,23 +48,32 @@ def download_from_oss(url, file_path):
     :param Union[LiteralString, str, bytes] file_path: 文件路径
     :return bool: 是否成功
     """
-    logging.info("download_from_oss: %s", url)
-    file_dir, file_name = os.path.split(file_path)
-    if file_dir:
-        if not os.path.exists(file_dir):
-            os.makedirs(file_dir)
-    max_retries = 5
-    retries = 0
-    while retries < max_retries:
-        try:
-            response = requests.get(url, timeout=15)  # 设置超时时间
-            with open(file_path, 'wb') as f:
-                f.write(response.content)
-            return True
-        except requests.exceptions.Timeout:
-            retries += 1
-            logger.warning(f"Download timed out. Retrying {retries}/{max_retries}...")
-        except Exception as e:
-            logger.warning(f"Download failed. Retrying {retries}/{max_retries}...")
-            retries += 1
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("download_from_oss") as span:
+        span.set_attribute("file.url", url)
+        span.set_attribute("file.path", file_path)
+        logging.info("download_from_oss: %s", url)
+        file_dir, file_name = os.path.split(file_path)
+        if file_dir:
+            if not os.path.exists(file_dir):
+                os.makedirs(file_dir)
+        max_retries = 5
+        retries = 0
+        while retries < max_retries:
+            try:
+                with tracer.start_as_current_span("download_from_oss.request") as req_span:
+                    req_span.set_attribute("http.method", "GET")
+                    req_span.set_attribute("http.url", url)
+                    response = requests.get(url, timeout=15)  # 设置超时时间
+                    req_span.set_attribute("http.status_code", response.status_code)
+                    with open(file_path, 'wb') as f:
+                        f.write(response.content)
+                    return True
+            except requests.exceptions.Timeout:
+                retries += 1
+                logger.warning(f"Download timed out. Retrying {retries}/{max_retries}...")
+            except Exception as e:
+                retries += 1
+                span.set_attribute("oss.error", str(e))
+                logger.warning(f"Download failed. Retrying {retries}/{max_retries}...")
     return False