Vtap renamed to agent

**Phenomenon and reproduction steps**

**Root cause and solution**

**Impactions**

**Test method**

**Affected branch(es)**

* main

**Checklist**

- [ ] Dependencies update required
- [ ] Common bug (similar problem in other repo)

Author: xiaochaoren1

app/app/application/l7_flow_tracing.py

@@ -44,7 +44,7 @@
         "resp_tcp_seq",
         "start_time_us",
         "end_time_us",
-        "vtap_id",
+        "agent_id",
         "tap_port",
         "tap_port_name",
         "tap_port_type",
@@ -402,7 +402,7 @@ async def trace_l7_flow(self,
                     if syscall_trace_id_request > 0 or syscall_trace_id_response > 0:
                         new_syscall_metas.add((
                             dataframe_flowmetas['_id'][index],
-                            dataframe_flowmetas['vtap_id'][index],
+                            dataframe_flowmetas['agent_id'][index],
                             dataframe_flowmetas['syscall_trace_id_request']
                             [index],
                             dataframe_flowmetas['syscall_trace_id_response']
@@ -486,7 +486,7 @@ async def trace_l7_flow(self,
                     id_to_related_tag = dict()
                     for index in new_flows.index:
                         _id = new_flows.at[index, '_id_str']
-                        vtap_id = new_flows.at[index, 'vtap_id']
+                        agent_id = new_flows.at[index, 'agent_id']
                         req_tcp_seq = new_flows.at[index, 'req_tcp_seq']
                         resp_tcp_seq = new_flows.at[index, 'resp_tcp_seq']
                         tap_side = new_flows.at[index, 'tap_side']
@@ -505,7 +505,7 @@ async def trace_l7_flow(self,
 
                         id_to_related_tag[_id] = {
                             '_id': _id,
-                            'vtap_id': vtap_id,
+                            'agent_id': agent_id,
                             'req_tcp_seq': req_tcp_seq,
                             'resp_tcp_seq': resp_tcp_seq,
                             'tap_side': tap_side,
@@ -630,7 +630,7 @@ async def query_flowmetas(self, time_filter: str,
             通过tcp_seq及流日志的时间追踪
     
         系统调用追踪信息：
-            vtap_id, syscall_trace_id_request, syscall_trace_id_response
+            agent_id, syscall_trace_id_request, syscall_trace_id_response
             通过eBPF获取到的coroutine_trace_id追踪
     
         主动注入的追踪信息：
@@ -640,7 +640,7 @@ async def query_flowmetas(self, time_filter: str,
         """
         sql = """SELECT 
         type, req_tcp_seq, resp_tcp_seq, toUnixTimestamp64Micro(start_time) AS start_time_us, toUnixTimestamp64Micro(end_time) AS end_time_us, 
-        vtap_id, syscall_trace_id_request, syscall_trace_id_response, span_id, parent_span_id, l7_protocol, 
+        agent_id, syscall_trace_id_request, syscall_trace_id_response, span_id, parent_span_id, l7_protocol, 
         trace_id, x_request_id_0, x_request_id_1, toString(_id) AS `_id_str`, tap_side, auto_instance_0, auto_instance_1
         FROM `l7_flow_log` 
         WHERE (({time_filter}) AND ({base_filter})) limit {l7_tracing_limit}
@@ -710,7 +710,7 @@ def set_all_relate(dataframe_flowmetas, related_map, network_delay_us):
         resp_tcp_seq = dataframe_flowmetas.at[index, 'resp_tcp_seq']
         tap_side = dataframe_flowmetas.at[index, 'tap_side']
         _id = dataframe_flowmetas.at[index, '_id']
-        vtap_id = dataframe_flowmetas.at[index, 'vtap_id']
+        agent_id = dataframe_flowmetas.at[index, 'agent_id']
         _type = dataframe_flowmetas.at[index, 'type']
         start_time_us = dataframe_flowmetas.at[index, 'start_time_us']
         end_time_us = dataframe_flowmetas.at[index, 'end_time_us']
@@ -726,7 +726,7 @@ def set_all_relate(dataframe_flowmetas, related_map, network_delay_us):
 
         id_to_related_tag[_id] = {
             '_id': _id,
-            'vtap_id': vtap_id,
+            'agent_id': agent_id,
             'req_tcp_seq': req_tcp_seq,
             'resp_tcp_seq': resp_tcp_seq,
             'tap_side': tap_side,
@@ -760,9 +760,9 @@ def set_all_relate(dataframe_flowmetas, related_map, network_delay_us):
         syscall_trace_id_response = dataframe_flowmetas.at[
             index, 'syscall_trace_id_response']
         _id = dataframe_flowmetas.at[index, '_id']
-        vtap_id = dataframe_flowmetas.at[index, 'vtap_id']
+        agent_id = dataframe_flowmetas.at[index, 'agent_id']
         if syscall_trace_id_request > 0 or syscall_trace_id_response > 0:
-            new_syscall_metas.add((_id, vtap_id, syscall_trace_id_request,
+            new_syscall_metas.add((_id, agent_id, syscall_trace_id_request,
                                    syscall_trace_id_response))
             if syscall_trace_id_request:
                 syscall_req_to_ids[syscall_trace_id_request].add(_id)
@@ -932,30 +932,30 @@ def set_relate(self, _ids, related_map, id_to_related_tag):
 class L7SyscallMeta:
     """
     系统调用追踪信息:
-        vtap_id, syscall_trace_id_request, syscall_trace_id_response, tap_side, start_time_us, end_time_us
+        agent_id, syscall_trace_id_request, syscall_trace_id_response, tap_side, start_time_us, end_time_us
     """
 
     def __init__(self, flow_metas: Tuple):
         self._id = flow_metas[0]
-        self.vtap_id = flow_metas[1]
+        self.agent_id = flow_metas[1]
         self.syscall_trace_id_request = flow_metas[2]
         self.syscall_trace_id_response = flow_metas[3]
 
     def __eq__(self, rhs):
-        return (self.vtap_id == rhs.vtap_id and self.syscall_trace_id_request
+        return (self.agent_id == rhs.agent_id and self.syscall_trace_id_request
                 == rhs.syscall_trace_id_request
                 and self.syscall_trace_id_response
                 == rhs.syscall_trace_id_response)
 
     def set_relate(self, _ids, related_map, id_to_related_tag):
         for _id in _ids:
             _id_df = id_to_related_tag[_id]['_id']
-            vtap_id_df = id_to_related_tag[_id]['vtap_id']
+            agent_id_df = id_to_related_tag[_id]['agent_id']
             syscall_trace_id_request_df = id_to_related_tag[_id][
                 'syscall_trace_id_request']
             syscall_trace_id_response_df = id_to_related_tag[_id][
                 'syscall_trace_id_response']
-            if _id_df == self._id or self.vtap_id != vtap_id_df:
+            if _id_df == self._id or self.agent_id != agent_id_df:
                 continue
             if self.syscall_trace_id_request > 0:
                 if self.syscall_trace_id_request == syscall_trace_id_request_df or self.syscall_trace_id_request == syscall_trace_id_response_df:
@@ -1096,8 +1096,8 @@ def sort_and_set_parent(self):
 
 class Service:
 
-    def __init__(self, vtap_id: int, process_id: int):
-        self.vtap_id = vtap_id
+    def __init__(self, agent_id: int, process_id: int):
+        self.agent_id = agent_id
         self.process_id = process_id
 
         self.direct_flows = []
@@ -1154,7 +1154,7 @@ def parent_set(self):
     def check_client_process_flow(self, flow: dict):
         """检查该flow是否与service有关联关系，s-p的时间范围需要覆盖c-p，否则拆分为两个service"""
         if self.process_id != flow["process_id_0"] \
-            or self.vtap_id != flow["vtap_id"]:
+            or self.agent_id != flow["agent_id"]:
             return False
         if self.start_time_us > flow["start_time_us"] \
             or self.end_time_us < flow["end_time_us"]:
@@ -1164,7 +1164,7 @@ def check_client_process_flow(self, flow: dict):
     def add_direct_flow(self, flow: dict):
         """direct_flow是指该服务直接接收到的，或直接发出的flow"""
         #assert (
-        #    self.vtap_id == flow.get('vtap_id')
+        #    self.agent_id == flow.get('agent_id')
         #    and self.process_id == flow.get('process_id')
         #)
         if flow['tap_side'] == TAP_SIDE_SERVER_PROCESS:
@@ -1246,7 +1246,7 @@ def merge_flow(flows: list, flow: dict) -> bool:
     if flow['type'] == L7_FLOW_TYPE_SESSION \
         and flow['tap_side'] not in [TAP_SIDE_SERVER_PROCESS, TAP_SIDE_CLIENT_PROCESS]:
         return False
-    # vtap_id, l7_protocol, flow_id, request_id
+    # agent_id, l7_protocol, flow_id, request_id
     for i in range(len(flows)):
         if flow['_id'] == flows[i]['_id']:
             continue
@@ -1281,7 +1281,7 @@ def merge_flow(flows: list, flow: dict) -> bool:
         if not request_flow or not response_flow:
             continue
         for key in [
-                'vtap_id', 'tap_port', 'tap_port_type', 'l7_protocol',
+                'agent_id', 'tap_port', 'tap_port_type', 'l7_protocol',
                 'request_id', 'tap_side'
         ]:
             if _get_df_key(request_flow, key) != _get_df_key(
@@ -1428,37 +1428,37 @@ def sort_all_flows(dataframe_flows: DataFrame, network_delay_us: int,
                     f"{id_map[_id]}-{','.join(related_types)}-{_id}")
         flow["related_ids"] = list(related_ids)
 
-    # 从Flow中提取Service：一个<vtap_id, local_process_id>二元组认为是一个Service。
+    # 从Flow中提取Service：一个<agent_id, local_process_id>二元组认为是一个Service。
     service_map = defaultdict(Service)
     for flow in syscall_flows:
         if flow['tap_side'] != TAP_SIDE_SERVER_PROCESS:
             continue
         local_process_id = flow['process_id_1']
-        vtap_id = flow['vtap_id']
-        if (vtap_id, local_process_id, 0) not in service_map:
-            service = Service(vtap_id, local_process_id)
-            service_map[(vtap_id, local_process_id, 0)] = service
+        agent_id = flow['agent_id']
+        if (agent_id, local_process_id, 0) not in service_map:
+            service = Service(agent_id, local_process_id)
+            service_map[(agent_id, local_process_id, 0)] = service
             # Service直接接收或发送的Flows_
             service.add_direct_flow(flow)
         else:
             index = 0
             for key in service_map.keys():
-                if key[0] == vtap_id and key[1] == local_process_id:
+                if key[0] == agent_id and key[1] == local_process_id:
                     index += 1
-            service = Service(vtap_id, local_process_id)
-            service_map[(vtap_id, local_process_id, index)] = service
+            service = Service(agent_id, local_process_id)
+            service_map[(agent_id, local_process_id, index)] = service
             service.add_direct_flow(flow)
 
     for flow in syscall_flows:
         if flow['tap_side'] != TAP_SIDE_CLIENT_PROCESS:
             continue
         local_process_id = flow['process_id_0']
-        vtap_id = flow['vtap_id']
+        agent_id = flow['agent_id']
         index = 0
         max_start_time_service = None
-        if (vtap_id, local_process_id, 0) in service_map:
+        if (agent_id, local_process_id, 0) in service_map:
             for key, service in service_map.items():
-                if key[0] == vtap_id and key[1] == local_process_id:
+                if key[0] == agent_id and key[1] == local_process_id:
                     index += 1
                     if service.check_client_process_flow(flow):
                         if not max_start_time_service:
@@ -1470,8 +1470,8 @@ def sort_all_flows(dataframe_flows: DataFrame, network_delay_us: int,
                 max_start_time_service.add_direct_flow(flow)
                 continue
         # 没有attach到service上的flow生成一个新的service
-        service = Service(vtap_id, local_process_id)
-        service_map[(vtap_id, local_process_id, index)] = service
+        service = Service(agent_id, local_process_id)
+        service_map[(agent_id, local_process_id, index)] = service
         # Service直接接收或发送的Flow
         service.add_direct_flow(flow)
 
@@ -1976,8 +1976,8 @@ def _get_flow_dict(flow: DataFrame):
         flow.get("childs", []),
         "process_id":
         flow.get("process_id", None),
-        "vtap_id":
-        flow.get("vtap_id", None),
+        "agent_id":
+        flow.get("agent_id", None),
         "service_uid":
         flow.get("service_uid", None),
         "service_uname":
@@ -2078,18 +2078,18 @@ def network_flow_sort(traces):
         sorted_traces += local_rest_traces
     else:
         for trace in local_rest_traces:
-            vtap_index = -1
+            agent_index = -1
             for i, sorted_trace in enumerate(sorted_traces):
-                if vtap_index > 0 and sorted_trace['vtap_id'] != trace[
-                        'vtap_id']:
+                if agent_index > 0 and sorted_trace['agent_id'] != trace[
+                        'agent_id']:
                     break
-                if sorted_trace['vtap_id'] == trace['vtap_id']:
+                if sorted_trace['agent_id'] == trace['agent_id']:
                     if sorted_trace['start_time_us'] < trace['start_time_us']:
-                        vtap_index = i + 1
-                    elif vtap_index == -1:
-                        vtap_index = i
-            if vtap_index >= 0:
-                sorted_traces.insert(vtap_index, trace)
+                        agent_index = i + 1
+                    elif agent_index == -1:
+                        agent_index = i
+            if agent_index >= 0:
+                sorted_traces.insert(agent_index, trace)
             else:
                 for i, sorted_trace in enumerate(sorted_traces):
                     if trace['start_time_us'] < sorted_trace['start_time_us']:

app/app/application/tracing_completion.py

@@ -240,7 +240,7 @@ async def trace_l7_flow(self,
                     if syscall_trace_id_request > 0 or syscall_trace_id_response > 0:
                         new_syscall_metas.add((
                             dataframe_flowmetas['_id'][index],
-                            dataframe_flowmetas['vtap_id'][index],
+                            dataframe_flowmetas['agent_id'][index],
                             dataframe_flowmetas['syscall_trace_id_request']
                             [index],
                             dataframe_flowmetas['syscall_trace_id_response']
@@ -323,7 +323,7 @@ async def trace_l7_flow(self,
                     id_to_related_tag = dict()
                     for index in new_flows.index:
                         _id = new_flows.at[index, '_id_str']
-                        vtap_id = new_flows.at[index, 'vtap_id']
+                        agent_id = new_flows.at[index, 'agent_id']
                         req_tcp_seq = new_flows.at[index, 'req_tcp_seq']
                         resp_tcp_seq = new_flows.at[index, 'resp_tcp_seq']
                         tap_side = new_flows.at[index, 'tap_side']
@@ -342,7 +342,7 @@ async def trace_l7_flow(self,
 
                         id_to_related_tag[_id] = {
                             '_id': _id,
-                            'vtap_id': vtap_id,
+                            'agent_id': agent_id,
                             'req_tcp_seq': req_tcp_seq,
                             'resp_tcp_seq': resp_tcp_seq,
                             'tap_side': tap_side,