data_integrate.py

import logging
import datetime
import psutil
import os
import pandas as pd
import numpy as np
import gc
import json

from multiprocessing import Pool
from more_itertools import locate
from log import Logger
from os.path import dirname
from log_parsing import *
from alarm import *
import pdb
import re
import tqdm

import concurrent.futures


log_path = dirname(__file__) + '/log/' + str(datetime.datetime.now().strftime(
    '%Y-%m-%d')) + '_nezha.log'
logger = Logger(log_path, logging.DEBUG, __name__).getlog()


def get_logs_within_trace_map(trace_reader, log_reader, trace_id):
    """
    "Sequence": [
        "Trace": {
            "TraceID": XXXX
            "Spans": [
                {
                    SpanID:
                    ParentID:
                    Pod:
                    "events": [
                        {
                            timestamp: xxx
                            event: xxx
                        },
                        {
                            timestamp: xxx
                            event: xxx
                        }
                    ]
                }
            ]
        }
    ]
    """
    trace = {"traceid": trace_id, "spans": []}
    try:
        # find all span within a trace
        spans = trace_reader.loc[[trace_id], ['SpanID', 'ParentID',
                                              'PodName', 'StartTimeUnixNano', 'OperationName']]
        if len(spans['SpanID']) > 0:
            for span_index in range(len(spans['SpanID'])):
                span_id = spans['SpanID'][span_index]
                pod = spans['PodName'][span_index]

                # Add opration event first
                span = {"spanid": span_id,
                        "parentid": spans['ParentID'][span_index],
                        "pod": pod, "events": []}

                # record span start and end event
                start_event = log = spans['OperationName'][span_index] + "_start"
                end_event = log = spans['OperationName'][span_index] + "_end"
                startlog = {"timestamp": spans['StartTimeUnixNano'][span_index],
                            "event": log_parsing(start_event, pod=pod)}
                span["events"].append(startlog)
                endlog = {"timestamp": spans['EndTimeUnixNano'][span_index],
                          "event": log_parsing(end_event, pod=pod)}
                span["events"].append(endlog)

                try:
                    # Find all log within a span
                    logs = log_reader.loc[[span_id],
                                          ['TimeUnixNano', 'Log']]
                    if len(logs['TimeUnixNano']) > 0:
                        for log_index in range(len(logs['TimeUnixNano'])):
                            # Add log events
                            log = {"timestamp": logs['TimeUnixNano'][log_index],
                                   "event": log_parsing(log=logs['Log'][log_index], pod=pod)}
                            span["events"].append(log)
                except (Exception):
                    pass
                # sort event by event timestamp
                span['events'].sort(key=lambda k: (k.get('timestamp', 0)))
                trace["spans"].append(span)
                # sort span by span start timestamp
                trace['spans'].sort(key=lambda k: k['events'][0]['timestamp'])
    except (Exception):
        pass
    return trace


class Trace(object):
    def __init__(self, traceid):
        self.traceid = traceid
        self.spans = []

    def sort_spans(self):
        self.spans.sort(key=lambda k: (k.events[0].timestamp, 0))

    def append_spans(self, span):
        self.spans.append(span)

    def show_all_spans(self):
        print(self.traceid)
        for i in range(len(self.spans)):
            self.spans[i].show_all_events()


class Span(object):
    def __init__(self, spanid, parentid, pod):
        self.spanid = spanid
        self.parentid = parentid
        self.pod = pod
        self.events = []

    def sort_events(self):
        self.events.sort(key=lambda k: (k.timestamp, 0))

    def append_event(self, event):
        self.events.append(event)

    def new_timestamp(self):
        return self.events[len(self.events)-1].timestamp - 1

    def show_all_events(self):
        logger.info("%s,%s,%s", self.spanid, self.parentid, self.pod)
        for i in range(len(self.events)):
            self.events[i].show_event()
        logger.info("")


class Event(object):
    def __init__(self, event, pod, ns, timestamp=None, spanid=None, parentid=None):
        self.event = event
        self.pod = pod
        self.timestamp = timestamp
        self.spanid = spanid
        self.parentid = parentid
        self.ns = ns 

    def show_event(self):
        logger.info("%s, %s, %s, %s", self.timestamp, self.spanid, self.event,
                    from_id_to_template(self.event, self.ns))


class EventGraph():
    def __init__(self, log_template_miner):
        self.adjacency_list = {}
        self.node_list = set()
        self.pair_set = set()
        self.support_dict = {}
        self.log_template_miner = log_template_miner

    def add_edge(self, node1, node2):
        if node1 not in self.adjacency_list.keys():
            self.adjacency_list[node1] = []
        self.adjacency_list[node1].append(node2)
        self.node_list.add(node1.event)
        self.node_list.add(node2.event)

    def remove_edge(self, node1, node2):
        # print(node1.event, node2.event)
        # for span in self.adjacency_list[node1]:
        #     print(span.event)
        self.adjacency_list[node1].remove(node2)

    def print_adj_list(self):
        for key in self.adjacency_list.keys():
            print(f'node {key}: {self.adjacency_list[key]}')

    def show_graph(self):
        for key in self.adjacency_list.keys():
            logger.info("head:%s %s" %
                        (key.event, from_id_to_template(key.event, self.log_template_miner)))
            for item in self.adjacency_list[key]:
                logger.info("tail:%s %s" %
                            (item.event, from_id_to_template(item.event, self.log_template_miner)))
            logger.info("----")

    def get_deepth_pod(self, traget_event):
        pod = ""
        deepth = 0
        while True:
            flag = False
            for key in self.adjacency_list.keys():
                for item in self.adjacency_list[key]:
                    if traget_event == item.event:
                        traget_event = key.event
                        if deepth == 0:
                            pod = item.pod
                        flag = True
                        if "start" in from_id_to_template(key.event, self.log_template_miner) and "TraceID" not in from_id_to_template(key.event, self.log_template_miner):
                            deepth = deepth + 1
                        break
                if flag == True:
                    break
            if flag == False:
                break
        return deepth, pod

    def get_support(self):
        for key in self.adjacency_list.keys():
            for item in self.adjacency_list[key]:
                supprot_key = str(key.event) + "_" + str(item.event)
                if supprot_key not in self.pair_set:
                    self.support_dict[supprot_key] = 1
                    self.pair_set.add(supprot_key)
                else:
                    self.support_dict[supprot_key] += 1
        # logger.info("support_dict: %s" % self.support_dict)
        return self.support_dict


def get_events_within_trace(trace_reader, log_reader, trace_id, alarm_list, ns,log_template_miner):
    """
    func get_events_within_trace: get all metric alarm, log, span within a trace and transform to event
    :parameter
        trace_read - pd.csv_read(tracefile)
        log_reader - pd.csv_read(logfile)
        trace_id   - only one trace is processed at a time
        alarm_list - [{'pod': 'cartservice-579f59597d-wc2lz', 'alarm': [{'metric_type': 'CpuUsageRate(%)', 'alarm_flag': True}]
    :return
        trace class with all event (events in the same span was order by timestamp)
    """
    # trace = {"traceid": trace_id, "spans": []}
    trace = Trace("StartTraceId is %s" % trace_id)
    # logger.info(trace_id)
    log_span_id_list = log_reader.index.tolist()
    try:
        # find all span within a trace
        spans = trace_reader.loc[[trace_id], ['SpanID', 'ParentID',
                                              'PodName', 'StartTimeUnixNano', 'EndTimeUnixNano', 'OperationName']]
        # pdb.set_trace()
        # print(len(spans['SpanID']))
        if len(spans['SpanID']) > 0:
            # process span independentlt and order by timestamp
            for span_index in range(len(spans['SpanID'])):
                # span event
                span_id = spans['SpanID'].iloc[span_index]
                parent_id = spans['ParentID'].iloc[span_index]
                pod = spans['PodName'].iloc[span_index]
                # Add opration event first
                span = Span(span_id, parent_id, pod)
                # record span start and end event

                service = pod.rsplit('-', 1)[0]
                service = service.rsplit('-', 1)[0]

                start_event = service + " " + \
                    spans['OperationName'].iloc[span_index] + " start"
                end_event = service + " " + \
                    spans['OperationName'].iloc[span_index] + " end"

                span.append_event(Event(timestamp=np.ceil(spans['StartTimeUnixNano'].iloc[span_index]).astype(int),
                                        event=log_parsing(log=start_event, pod=pod, log_template_miner=log_template_miner), pod=pod, ns=ns, spanid=span_id, parentid=parent_id))

                end_timestamp = np.ceil(
                    spans['EndTimeUnixNano'].iloc[span_index]).astype(int)

                # log event
                try:
                    # pdb.set_trace()
                    if span_id in log_span_id_list:
                        logs = log_reader.loc[[span_id],
                                              ['TimeUnixNano', 'Log']]

                        if len(logs['TimeUnixNano']) > 0:
                            for log_index in range(len(logs['TimeUnixNano'])):
                                # Add log events
                                # logger.info(logs['Log'].iloc[log_index])
                                log = logs['Log'].iloc[log_index]
                                timestamp = np.ceil(
                                    logs['TimeUnixNano'].iloc[log_index]).astype(int)

                                # for log end after span
                                if timestamp - end_timestamp > 0:
                                    # logger.info(json.loads(log)['log'])
                                    end_timestamp = timestamp + 1

                                # if log_parsing(log=log, pod=pod) != histroy_id:
                                #     histroy_id = log_parsing(log=log, pod=pod)
                                span.append_event(Event(timestamp=timestamp,
                                                        event=log_parsing(log=log, pod=pod, log_template_miner=log_template_miner), pod=pod, ns=ns, spanid=span_id, parentid=parent_id))

                    # else:
                    # logger.debug("Spanid %s is not appear in log" % span_id)
                except Exception as e:
                    logger.error("Catch an exception:", e)
                    pass

                span.append_event(Event(timestamp=end_timestamp,
                                        event=log_parsing(log=end_event, pod=pod, log_template_miner=log_template_miner), pod=pod, ns=ns,spanid=span_id, parentid=parent_id))
                # alarm event
                # hipster
                if ns == "hipster":
                    if len(span.events) > 2:
                        # if len(span.events) >= 2:
                        # do not add alarm for client span
                        for i in range(len(alarm_list)):
                            alarm_dict = alarm_list[i]
                            if alarm_dict["pod"] == span.pod:
                                # logger.info(
                                #     "%s, %s", alarm_dict["pod"], alarm_dict["alarm"])
                                for index in range(len(alarm_dict["alarm"])):
                                    span.append_event(Event(timestamp=np.ceil(spans['StartTimeUnixNano'].iloc[span_index]).astype(int) + index + 1,
                                                            event=log_parsing(log=alarm_dict["alarm"][index]["metric_type"], pod="alarm",log_template_miner=log_template_miner), pod=pod, ns=ns,spanid=span_id, parentid=parent_id))
                                # only add alarm event for the first span group
                                # alarm_list.pop(i)
                                break
                elif ns == "ts":
                    for i in range(len(alarm_list)):
                        alarm_dict = alarm_list[i]
                        if alarm_dict["pod"] == span.pod:
                            # logger.info(
                            #     "%s, %s", alarm_dict["pod"], alarm_dict["alarm"])
                            for index in range(len(alarm_dict["alarm"])):
                                span.append_event(Event(timestamp=np.ceil(spans['StartTimeUnixNano'].iloc[span_index]).astype(int) + index + 1,
                                                        event=log_parsing(log=alarm_dict["alarm"][index]["metric_type"], pod="alarm",log_template_miner=log_template_miner), pod=pod, ns=ns,spanid=span_id, parentid=parent_id))
                            # only add alarm event for the first span group
                            # alarm_list.pop(i)
                            break                    

                # sort event by event timestamp
                span.sort_events()

                trace.append_spans(span)

            # sort span by span start timestamp
            trace.sort_spans()
    except Exception as e:
        pass
        # logger.error("Catch an exception: %s", e)

    return trace


def generate_event_chain(alarm_list, trace,log_template_miner):
    """
    func generate_event_chain:  integrate log and trace to event chain (Discarded)
    :parameter
        trace - trace including spans with all log from get_logs_within_trace
    :return
        pod_event_sequences -  e.g.,  [['frontend-579b9bff58-hfpmg_4', 'frontend-579b9bff58-hfpmg_4'], ['frontend-579b9bff58-hfpmg_4']]
        event_sequences - e.g., [[1,2,3], [2,3,4]]
    """
    event_chain = []
    pod_event_chain_list = []  # [pod_1, pod_2]
    event_chain_list = []  # [1, 2, 3]
    # trace.show_all_spans()
    for span in trace.spans:
        if span.parentid == "root":
            for event in span.events:
                event_chain.append(event)

        elif len(span.events) == 2:
            # for client span
            # find all event index of parent span
            parent_index1 = list(locate(event_chain, lambda event: event.spanid ==
                                        span.parentid and event.pod == span.pod))
            # find all event index of spans having the same parent span
            parent_index2 = list(locate(event_chain, lambda event: event.parentid ==
                                        span.parentid and event.pod == span.pod))
            parent_index = parent_index2 + parent_index1
            # reverse order
            parent_index.sort(reverse=True)
            # reverse event span to insert the late event first
            span.events.reverse()
            for event in span.events:
                for index in parent_index:
                    if event.timestamp > event_chain[index].timestamp:
                        event_chain.insert(index+1, event)
                        break

        else:
            # # find all event index of parent span
            parent_index1 = list(locate(event_chain, lambda event: event.spanid ==
                                        span.parentid))
            parent_index2 = list(locate(event_chain, lambda event: event.parentid ==
                                        span.parentid and event.pod == span.pod))
            parent_index = parent_index2 + parent_index1
            parent_index.sort(reverse=True)
            span.events.reverse()
            for event in span.events:
                for index in parent_index:
                    if event.timestamp > event_chain[index].timestamp:
                        event_chain.insert(index+1, event)
                        break

    # for index in range(1, len(event_chain)):
    #     if event_chain[index].timestamp < event_chain[index-1].timestamp:
    #         event_chain[index-1].show_event()
    #         event_chain[index].show_event()
    #         print()

    # chain event class to str: pod_event
    for index in range(len(event_chain)):
        pod_event_chain_list.append(event_chain[index].pod +
                                    "_" + str(event_chain[index].event))
        event_chain_list.append(event_chain[index].event)

    # insert alarm before the first event belonging to the pod has alarm
    if len(alarm_list) > 0:
        for pod_alarm in alarm_list:
            pod = pod_alarm['pod']
            for index in range(len(pod_event_chain_list)):
                if re.search(pod, pod_event_chain_list[index]):
                    for alarm in pod_alarm['alarm']:
                        event = log_parsing(
                            log=alarm['metric_type'], pod="alarm", log_template_miner=log_template_miner)
                        pod_event = pod + str(event)
                        pod_event_chain_list.insert(index, pod_event)
                        event_chain_list.insert(index, event)
                    break

    return pod_event_chain_list, event_chain_list


def generate_event_graph(trace, log_template_miner):
    """
    func generate_event_graph: integrate events of different span to graph
    :parameter
        trace - trace including spans with all event from get_events_within_trace
    :return
        event_graph -
    """
    event_graph = EventGraph(log_template_miner)
    # trace.show_all_spans()

    for span in trace.spans:
        # add edge in the span group
        for index in range(1, len(span.events)):
            event_graph.add_edge(
                span.events[index-1], span.events[index])

    for span in trace.spans:
        # add relation from parent to child
        # if span.parentid != "root":
        for parent_span in trace.spans:
            if parent_span.spanid == span.parentid:
                if parent_span.pod == span.pod:
                    # if in the same pod, insert based on timestamp
                    start_timestamp = span.events[0].timestamp

                    for index in range(1, len(parent_span.events)):
                        if parent_span.events[index].timestamp > start_timestamp:
                            event_graph.add_edge(
                                parent_span.events[index-1], span.events[0])
                            break

                else:
                    # if not in the same pod, insert after the first span of parent group
                    event_graph.add_edge(
                        parent_span.events[0], span.events[0])
                break
    # event_graph.show_graph()
    # event_graph.get_support()
    # print(event_graph.support_dict)
    return event_graph


def data_integrate(trace_file, trace_id_file, log_file, alarm_list, ns,log_template_miner):
    """
    func data_integrate: integrate multimodle data to event graph
    :parameter
        trace_file
        trace_id_file
        log_file
        alarm_list
        ns
    :return
        list of event graph
    """
    logger.info(alarm_list)
    # alarm_list = []
    trace_id_reader = pd.read_csv(
        trace_id_file, index_col=False, header=None, engine='c')
    trace_reader = pd.read_csv(
        trace_file, index_col='TraceID', usecols=['TraceID', 'SpanID', 'ParentID', 'PodName', 'StartTimeUnixNano', 'EndTimeUnixNano', 'OperationName'],engine='c')
    log_reader = pd.read_csv(log_file, index_col='SpanID', usecols=[
        'TimeUnixNano', 'SpanID', 'Log'], engine='c')
    log_sequences = []
    traces = []
    # print(datetime.datetime.now())
    event_graphs = []
    # pool = Pool(200)

    # traces = [pool.apply_async(get_logs_within_trace_map, args=(
    #     trace_reader, log_reader, trace_id_reader[0][i],)) for i in range(len(trace_id_reader[0]))]

    # for test
    # trace = get_events_within_trace(trace_reader, log_reader,
    #                                 trace_id_reader[0][3], alarm_list)
    # graph = generate_event_graph(trace)
    # event_graphs.append(graph)

    with concurrent.futures.ProcessPoolExecutor(max_workers=64) as executor1:
        futures1 = {executor1.submit(
            get_events_within_trace, trace_reader, log_reader, traceid, alarm_list, ns,log_template_miner) for traceid in trace_id_reader[0]}

        for future1 in concurrent.futures.as_completed(futures1):
            trace = future1.result()
            if trace is not None:
                log_sequences.append(trace)
        executor1.shutdown()

    with concurrent.futures.ProcessPoolExecutor(max_workers=64) as executor2:
        futures2 = {executor2.submit(
            generate_event_graph, trace,log_template_miner) for trace in log_sequences}

        for future2 in concurrent.futures.as_completed(futures2):
            graph = future2.result()
            if graph is not None:
                event_graphs.append(graph)
        executor2.shutdown()

    # for running
    # traces = [pool.apply_async(get_events_within_trace, args=(
    #     trace_reader, log_reader, traceid, alarm_list, ns,)) for traceid in trace_id_reader[0]]
    # for trace in traces:
    #     log_sequences.append(trace.get())

    # graphs = [pool.apply_async(
    #     generate_event_graph, args=(trace,ns,)) for trace in log_sequences]
    # for graph in graphs:
    #     event_graphs.append(graph.get())

    for graph in event_graphs:
        # logger.info("Start show graph")
        # graph.show_graph()
        graph.get_support()
        # graph.get_deepth(107)

    # del trace_id_reader
    # del trace_reader
    # del log_reader

    # pool.close()
    # pool.join()

    # events[0][0].show_event()
    # print(datetime.datetime.now())
    logger.info("Data Integrate Complete!")
    return event_graphs


if __name__ == '__main__':
    date = "2023-01-29"
    hour_min = "08_50"
    construction_data_path = "./construct_data"
    trace_file = construction_data_path + "/" + \
        date + "/trace/" + hour_min + "_trace.csv"
    trace_id_file = construction_data_path + "/" + date + \
        "/traceid/" + hour_min + "_traceid.csv"
    log_file = construction_data_path + "/" + date + \
        "/log/" + hour_min + "_log.csv"
    
    ns = "ts"

    alarm_list = []
    event_graphs = data_integrate(
        trace_file, trace_id_file, log_file, alarm_list, ns)