一、CANN 日志基础
1.1 日志级别
CANN 日志级别 (从低到高): DEBUG → 调试信息，开发阶段使用 INFO → 一般信息，默认级别 WARNING → 警告信息，不影响运行 ERROR → 错误信息，功能受损 FATAL → 致命错误，程序终止
1.2 基础配置
# 环境变量配置exportASCEND_LOG_LEVEL=3# INFO 级别exportASCEND_LOG_PATH=/var/log/ascendexportASCEND_LOG_FILE=ascend.logexportASCEND_LOG_FILE_SIZE=100# MBexportASCEND_LOG_FILE_NUM=10# 保留文件数# 运行时查看日志tail-f/var/log/ascend/ascend.log
1.3 Python 日志接口
importlogging# 获取 CANN 日志器logger=logging.getLogger('ascend')# 配置日志格式handler=logging.StreamHandler()formatter=logging.Formatter('[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',datefmt='%Y-%m-%d %H:%M:%S')handler.setFormatter(formatter)logger.addHandler(handler)logger.setLevel(logging.INFO)# 使用日志logger.info("模型加载开始")logger.warning("显存使用率超过 80%")logger.error("模型转换失败")
二、自定义日志
2.1 推理日志装饰器
importfunctoolsimporttimeimportloggingdeflog_inference(func):"""推理日志装饰器"""@functools.wraps(func)defwrapper(*args,**kwargs):logger=logging.getLogger('inference')start_time=time.time()logger.info(f"开始推理:{func.__name__}")try:result=func(*args,**kwargs)latency=time.time()-start_time logger.info(f"推理完成:{func.__name__}"f"耗时={latency:.3f}s "f"输入形状={args[0].shapeifhasattr(args[0],'shape')else'N/A'}")returnresultexceptExceptionase:latency=time.time()-start_time logger.error(f"推理失败:{func.__name__}"f"耗时={latency:.3f}s "f"错误={str(e)}")raisereturnwrapper# 使用示例@log_inferencedefpredict(model,input_data):returnmodel(input_data)
2.2 性能追踪日志
classPerformanceTracer:def__init__(self,log_file='performance.jsonl'):self.log_file=log_file self.logger=logging.getLogger('performance')deftrace(self,operation_name):"""性能追踪上下文管理器"""returnTraceContext(self,operation_name)defrecord(self,operation_name,duration,metadata=None):"""记录性能数据"""record={'timestamp':time.time(),'operation':operation_name,'duration_ms':duration*1000,'metadata':metadataor{}}# 写入 JSONL 文件withopen(self.log_file,'a')asf:f.write(json.dumps(record)+' ')# 同时输出到日志self.logger.info(f"性能:{operation_name}"f"耗时={duration*1000:.2f}ms "f"元数据={metadata}")classTraceContext:def__init__(self,tracer,operation_name):self.tracer=tracer self.operation_name=operation_name self.start_time=Nonedef__enter__(self):self.start_time=time.time()returnselfdef__exit__(self,exc_type,exc_val,exc_tb):duration=time.time()-self.start_time metadata={}ifexc_type:metadata['error']=str(exc_val)self.tracer.record(self.operation_name,duration,metadata)returnFalse# 使用示例tracer=PerformanceTracer()withtracer.trace("model_load"):model=load_model("model.om")withtracer.trace("preprocess"):input_data=preprocess(raw_data)withtracer.trace("inference"):output=model(input_data)withtracer.trace("postprocess"):result=postprocess(output)
三、分布式日志收集
3.1 日志聚合
importloggingimportlogging.handlersimportsocketclassDistributedLogHandler(logging.Handler):"""分布式日志处理器"""def__init__(self,host,port,node_id):super().__init__()self.host=host self.port=port self.node_id=node_id self.socket=socket.socket(socket.AF_INET,socket.SOCK_STREAM)self.socket.connect((host,port))defemit(self,record):"""发送日志到中心服务器"""try:log_entry={'node_id':self.node_id,'timestamp':record.created,'level':record.levelname,'message':self.format(record),'module':record.module,'function':record.funcName,'line':record.lineno}message=json.dumps(log_entry).encode('utf-8')self.socket.send(message)exceptExceptionase:self.handleError(record)# 配置分布式日志defsetup_distributed_logging(node_id,log_server='192.168.1.100',port=5000):"""配置分布式日志"""logger=logging.getLogger()logger.setLevel(logging.INFO)# 控制台输出console_handler=logging.StreamHandler()console_handler.setFormatter(logging.Formatter('[%(asctime)s] [%(node_id)s] [%(levelname)s] %(message)s'))logger.addHandler(console_handler)# 分布式输出distributed_handler=DistributedLogHandler(log_server,port,node_id)distributed_handler.setFormatter(logging.Formatter('[%(asctime)s] [%(node_id)s] [%(levelname)s] [%(module)s:%(funcName)s:%(lineno)d] %(message)s'))logger.addHandler(distributed_handler)returnlogger
3.2 ELK 日志栈
# docker-compose-elk.ymlversion:'3.8'services:elasticsearch:image:docker.elastic.co/elasticsearch/elasticsearch:8.10.0environment:-discovery.type=single-node-xpack.security.enabled=falseports:-"9200:9200"volumes:-esdata:/usr/share/elasticsearch/datalogstash:image:docker.elastic.co/logstash/logstash:8.10.0volumes:-./logstash.conf:/usr/share/logstash/pipeline/logstash.confports:-"5000:5000"depends_on:-elasticsearchkibana:image:docker.elastic.co/kibana/kibana:8.10.0ports:-"5601:5601"depends_on:-elasticsearchvolumes:esdata:
# logstash.confinput{tcp{port=>5000codec=>json_lines}}filter{# 添加字段mutate{add_field=>{"service"=>"cann-inference"}}# 解析时间戳date{match=>["timestamp","UNIX_MS"]}}output{elasticsearch{hosts=>["elasticsearch:9200"]index=>"cann-logs-%{+YYYY.MM.dd}"}}
四、基于日志的性能分析
4.1 日志分析脚本
importjsonimportpandasaspdfromcollectionsimportdefaultdictclassLogAnalyzer:def__init__(self,log_file):self.log_file=log_file self.records=self._parse_log()def_parse_log(self):"""解析日志文件"""records=[]withopen(self.log_file,'r')asf:forlineinf:try:record=json.loads(line.strip())records.append(record)exceptjson.JSONDecodeError:continuereturnrecordsdefperformance_summary(self):"""性能摘要"""df=pd.DataFrame(self.records)if'duration_ms'notindf.columns:print("无性能数据")returnsummary=df.groupby('operation')['duration_ms'].agg(['count','mean','std','min','max',lambdax:x.quantile(0.5),lambdax:x.quantile(0.95),lambdax:x.quantile(0.99)])summary.columns=['次数','平均(ms)','标准差','最小(ms)','最大(ms)','P50(ms)','P95(ms)','P99(ms)']print(" 性能摘要:")print(summary.to_string())returnsummarydefbottleneck_analysis(self):"""瓶颈分析"""df=pd.DataFrame(self.records)if'duration_ms'notindf.columns:return# 按操作统计operation_stats=df.groupby('operation')['duration_ms'].agg(['mean','sum','count'])operation_stats=operation_stats.sort_values('sum',ascending=False)print(" 瓶颈分析(按总耗时排序):")print(operation_stats.to_string())# 识别瓶颈total_time=operation_stats['sum'].sum()cumulative=0bottlenecks=[]forop,rowinoperation_stats.iterrows():cumulative+=row['sum']percentage=cumulative/total_time*100bottlenecks.append((op,row['sum'],percentage))ifpercentage>=80:# 80/20 法则breakprint(f" 前{len(bottlenecks)}个操作占总耗时80%:")forop,total,pctinbottlenecks:print(f"{op}:{total:.2f}ms ({pct:.1f}%)")# 使用示例analyzer=LogAnalyzer('performance.jsonl')analyzer.performance_summary()analyzer.bottleneck_analysis()
4.2 实时监控
classRealtimeLogMonitor:def__init__(self,log_file):self.log_file=log_file self.metrics=defaultdict(list)self.lock=threading.Lock()defstart(self):"""启动实时监控"""self.thread=threading.Thread(target=self._monitor_loop,daemon=True)self.thread.start()def_monitor_loop(self):"""监控循环"""withopen(self.log_file,'r')asf:f.seek(0,2)# 跳到文件末尾whileTrue:line=f.readline()ifnotline:time.sleep(0.1)continuetry:record=json.loads(line.strip())self._process_record(record)exceptjson.JSONDecodeError:continuedef_process_record(self,record):"""处理日志记录"""withself.lock:if'duration_ms'inrecord:op=record.get('operation','unknown')self.metrics[op].append(record['duration_ms'])# 保持最近 1000 条iflen(self.metrics[op])>1000:self.metrics[op]=self.metrics[op][-1000:]defget_current_metrics(self):"""获取当前指标"""withself.lock:metrics={}forop,durationsinself.metrics.items():ifdurations:metrics[op]={'count':len(durations),'avg':sum(durations)/len(durations),'p99':sorted(durations)[int(len(durations)*0.99)]}returnmetrics# 使用示例monitor=RealtimeLogMonitor('performance.jsonl')monitor.start()# 定期获取指标whileTrue:time.sleep(10)metrics=monitor.get_current_metrics()print(json.dumps(metrics,indent=2))
五、常见问题
| 问题 | 原因 | 解决方案 |
|---|
| 日志文件过大 | 日志级别太低 | 调整为 INFO 或 WARNING |
| 日志缺失 | 未正确配置路径 | 检查 ASCEND_LOG_PATH |
| 性能日志格式错误 | JSON 解析失败 | 检查日志格式 |
| 分布式日志延迟 | 网络问题 | 增加缓冲区、异步发送 |
| 日志文件损坏 | 异常终止 | 使用日志轮转 |
相关仓库
- CANN- 昇腾异构计算架构 https://atomgit.com/cann
- ge- 图引擎 https://atomgit.com/cann/ge
- cann-recipes-infer- 推理配方 https://atomgit.com/cann/ops-nn
- runtime- 运行时 https://atomgit.com/cann/runtime
- cann-samples- 示例代码 https://atomgit.com/cann-samples
- hccl- 集合通信库 https://atomgit.com/cann/hccl
