text: # Copyright 2019 Google LLC
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     https://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """Processes transperf outputs including pcap files and kernel log files.
  """
  
  import csv
  import logging
  import re
  
  from scapy.all import PcapReader
  
  LOG = logging.getLogger('transperf/outparser')
  
  
  class Visitor(object):
      """The abstract base class for all the classes that process output files.
  
      Visitors are added to transperf to process logs and pcap files in one pass,
      *hopefully* with O(1) memory.
  
      Visitor methods are called in the following sequence:
      1) begin()
      2) visit_conn()
      3) visit_packet()
      4) visit_klog()
      5) end()
      """
  
      def begin(self, exp, exp_dir, rcv_ip):
          """Called when the visitor should start a new experiment.
  
          Args:
              exp: The experiment object.
              exp_dir: The experiment output directory.
              rcv_ip: The receiver's IP address.
          """
          pass
  
      def end(self):
          """Called when all the output entries are passed to the visitor."""
          pass
  
      def visit_conn(self, ip, port, tool, cc, params, start, dur, tput):
          """Called for each connection.
  
          Args:
              ip: The ip address of the connection.
              port: The port of the connection.
              tool: The tool used in the experiment.
              cc: The congestion control algorithm.
              params: Parameters used for cc.
              start: The relative start time in seconds.
              dur: The duration of the connection in seconds.
              tput: The throughput reported by the benchmarking
                    application (e.g., netperf).
          """
          pass
  
      def visit_klog(self, time, line, match):
          """Called for a new klog line.
  
          The log lines of each connection are sorted by time, but the lines of
          different connections can interleave.
  
          Args:
              time: Time of the log entry relative to the start time of the
                    experiment.
              line: The raw content of the log file.
              match: The dictionary of all grouped regex matches.
          """
          pass
  
      def visit_strmr_log(self, time, pline):
          """Called for a new streamer log line.
  
          The log lines of each connection are sorted by time.
  
          Args:
              time: Time of the log entry relative to the start time of the
                    experiment.
              pline: The dictionary of all relvant parsed fields of a log line.
          """
          pass
  
      def visit_ss_log(self, time, data):
          """Called for a new ss log entry.
  
          The log entries are sorted by time.
  
          Args:
              time: The time of the log entry when run ss command.
              data: The dictionary of all relvant parsed fields of a log line.
          """
          pass
  
      def visit_packet(self, time, packet):
          """Called when there is a new packet available to be processed.
  
          The packets of each connection are sorted by time but packets of
          different connections can interleave..
  
          Args:
              time: Captured time relative to the start time of the experiment.
              packet: The packet parsed by scapy.
          """
          pass
  
      def visit_metric(self, metric):
          """Called when a metric is available to be processed.
  
          Args:
              metric: The metric of type metric.Metric.
          """
          pass
  
  
  class SsLog(object):
      """Parses ss logs and provides the flows of the experiment.
  
      Attributes:
          __readers: The ss log file readers.
          __entries: The most recent read entry from each log file. We keep this
                     list to make sure the entries are yielded sorted by time.
      """
  
      def __init__(self, log_paths):
          self.__readers = [open(path) for path in log_paths]
          self.__times = [0] * len(log_paths)
          self.__entries = [None] * len(log_paths)
  
      def __read_sslog(self, i):
          """Read the next entry in file.
  
          Args:
              i: The index of the file reader.
  
          Returns:
              The next entry in file f. None if there is no entry.
          """
          f = self.__readers[i]
          if not f:
              return None
          time = self.__times[i]
          line = f.readline()
          if not line:
              return None
          while line.startswith('# '):
              self.__times[i] = time = float(line[2:])
              f.readline()
              line = f.readline()
              if not line:
                  return None
          data = {}
          port = line.strip()
          port = int(port[port.rfind(':') + 1:])
          data['port'] = port
          line = f.readline()
          if not line:
              return None
          stat = line.strip().split()
          for item in stat:
              if item.startswith('bytes_acked:'):
                  data['bytes_acked'] = int(item[item.rfind(':') + 1:])
              elif item.startswith('retrans:'):
                  data['retrans'] = int(item[item.rfind('/') + 1:])
              elif item.startswith('data_segs_out:'):
                  data['data_segs_out'] = int(item[item.rfind(':') + 1:])
              elif item.startswith('rtt:'):
                  data['rtt'] = (
                      float(item[item.find(':') + 1:item.rfind('/')]) / 1000
                  )
              elif item.startswith('unacked:'):
                  data['unacked'] = int(item[item.find(':') + 1:])
          return time, data
  
      def __next_entry(self):
          """Returns the next entry ordered by time.
  
          Returns:
              The next entry. None if there is no entry.
          """
          min_time = -1
          min_index = -1
          for i, entry in enumerate(self.__entries):
              # If the reader has finished reading entries, check the next slot.
              if not self.__readers[i]:
                  continue
  
              # Fill the holes.
              if not entry:
                  entry = self.__read_sslog(i)
                  self.__entries[i] = entry
  
              # If entry is not set, it means that there is no entry in the
              # reader. So, we can remove the reader.
              if not entry:
                  self.__readers[i] = None
                  continue
              entry_time = entry[0]
              if min_index == -1 or entry_time < min_time:
                  min_index = i
                  min_time = entry_time
  
          if min_index == -1:
              return None
  
          entry = self.__entries[min_index]
          self.__entries[min_index] = None
          return entry
  
      def entries(self):
          """Entries stored in the ss log files.
  
          Yields:
              A tuple in the form of (relative time in sec, entry).
          """
          min_time = -1
          while True:
              entry = self.__next_entry()
              if not entry:
                  break
  
              if min_time == -1:
                  min_time = entry[0]
  
              yield (entry[0] - min_time, entry[1])
  
  
  class Pcap(object):
      """Parses pcap files and provides the flows of the experiment.
  
      Attributes:
          __readers: The pcap readers.
          __packets: The most recent read packet from each pcap file. We keep this
                     list to make sure the packets are yielded sorted by time.
      """
  
      def __init__(self, pcap_paths):
          self.__readers = [PcapReader(path) for path in pcap_paths]
          self.__packets = [None] * len(pcap_paths)
  
      def __next_packet(self):
          """Returns the next packet ordered by time.
  
          Returns:
              The next packet. None if there is no packet.
          """
          min_time = -1
          min_index = -1
          for i, pkt in enumerate(self.__packets):
              # If the reader has finished reading packets, check the next slot.
              if not self.__readers[i]:
                  continue
  
              # Fill the holes.
              if not pkt:
                  self.__packets[i] = pkt = self.__readers[i].read_packet()
  
              # If pkt is not set, it means that there is no packet in the reader.
              # So, we can remove the reader.
              if not pkt:
                  self.__readers[i] = None
                  continue
  
              if min_index == -1 or pkt.time < min_time:
                  min_index = i
                  min_time = pkt.time
  
          if min_index == -1:
              return None
  
          pkt = self.__packets[min_index]
          self.__packets[min_index] = None
          return pkt
  
      def packets(self):
          """Packets stored in the pcap files.
  
          Yields:
              A tuple in the form of (relative time in sec, raw packet, ip, tcp).
          """
          min_time = -1
          while True:
              pkt = self.__next_packet()
              if not pkt:
                  break
  
              if min_time == -1:
                  min_time = pkt.time
  
              yield (pkt.time - min_time, pkt)
  
  
  # These are regular expressions to parse congestion control output in
  # kern-debug.log.
  _LOG_PATTERNS = [
      # BBR:
      re.compile((
          r'\w+\s+\d+\s+\d{2}:\d{2}:\d{2}\s+(\w[\w\d\-]+)\s+kernel:\s+'
          r'\[\s*(?P<ts>[\d\.]+)\] BBR '
          r'(?P<ip>(\d{1,3}\.){3}\d{1,3}):(?P<port>\d{1,6})\s+'
          r'(?P<ack>[\d,]+):(?P<fack>\d+)\s+'
          r'(?P<castate>\S)\s+(?P<mode>\S)\s+'
          r'(?P<snd_cwnd>\d+)\s+'
          r'br\s+(?P<extra_acked>\d+)\s+'
          r'cr\s+(?P<crtt>-?\d+)\s+'
          r'rtt\s+(?P<rtt>-?\d+)\s+'
          r'd\s+(?P<rs_delivered>-?\d+)\s+'
          r'i\s+(?P<interval_us>-?\d+)\s+'
          r'mrtt\s+(?P<mrtt>-?\d+)\s+'
          r'(?P<rs_app_limited>\S)bw\s+(?P<sample_bw>\d+)\s+'
          r'bw\s+(?P<bw>\d+)\s+'
          r'lb\s+(?P<unused1>\d+)\s+'
          r'ib\s+(?P<interval_bw>\d+)\s+'
          r'qb\s+(?P<pacing_bw>\d+)\s+'
          r'a\s+(?P<acked>\d+)\s+'
          r'if\s+(?P<inflight>\d+)\s+'
          r'(?P<unused2>\S)\s+'
          r'(?P<round_start>\S)\s+'
          r'dl\s+(?P<tp_delivered>\d+)\s+'
          r'l\s+(?P<tp_loss>\d+)\s+'
          r'al\s+(?P<tp_app_limited>\d+)\s+'
          r'#\s+(?P<unused3>\d+)\s+'
          r't\s+(?P<targetcw>\d+)\s+'
          r'(?P<reord_seen>r|\.)\s+'
          r'(?P<prev_ca_state>O|D|C|R|L)\s+'
          r'lr\s+(?P<lr_x1000>-?\d+)\s+'
          r'er\s+(?P<ecn_x1000>-?\d+)\s+'
          r'ea\s+(?P<ecn_alpha_x1000>-?\d+)\s+'
          r'bwl\s+(?P<bw_lo>-?\d+)\s+'
          r'il\s+(?P<inflight_lo>-?\d+)\s+'
          r'ih\s+(?P<inflight_hi>-?\d+)\s+'
          r'c\s+(?P<bw_probe_up_cnt>-?\d+)\s+'
          r'v\s+(?P<version>-?\d+)\s+'
          r'(?P<debug_event>[\S])\s+'
          r'(?P<cycle_idx>\d+)\s+'
          r'(?P<ack_phase>I|R|B|F|A)\s+'
          r'(?P<bw_probe_samples>Y|N)'
      )),
  ]
  
  
  class KernLog(object):
      """Parses kern-debug.log files.
  
      Attributes:
          __log_paths: The paths of kernel log files.
      """
  
      def __init__(self, log_paths):
          self.__log_paths = log_paths
  
      def lines(self):
          """Yields a tuple for each log entry.
  
          Yields:
              Tuples in the form of: (timestamp in sec, raw line, parsed line)
          """
          min_ts = {}
          for path in self.__log_paths:
              f = open(path)
              for l in f:
                  # All log patterns must have "ts" and "port" fields.
                  m = None
                  for p in _LOG_PATTERNS:
                      m = p.match(l.strip())
                      if m:
                          break
                  if not m:
                      LOG.debug('cannot match log line: %s', l)
                      continue
  
                  mdict = m.groupdict()
                  if 'ts' not in mdict or 'port' not in mdict:
                      LOG.debug('no port or timestamp in log line: %s', l)
                      continue
  
                  ts = float(mdict['ts'])
  
                  # Make timestamps relative to the timestamp of the first
                  # entry of this port in the log file.
                  port = mdict['port']
                  if port not in min_ts:
                      min_ts[port] = ts
                      ts = 0
                  else:
                      ts -= min_ts[port]
  
                  yield (ts, l, m.groupdict())
  
  
  class ConnInfo(object):
      """Parses the exp_dir/conn.info file.
  
      This file is dumped by the sender and includes a line per connection.
      """
  
      def __init__(self, cinfo_files):
          self.__port_infos = {}
          for f in cinfo_files:
              lines = open(f).readlines()
              for l in lines:
                  l = l.strip()
                  port, conn_info = l.split('=', 1)
                  self.__port_infos[int(port)] = conn_info.split(',', 6)
  
      def conn_info(self, port):
          """Connection information of the given port."""
          return self.__port_infos[port]
  
      def ports(self):
          """Ports that exist in the conn.info files."""
          return self.__port_infos.keys()
  
  
  class RecvInfo(object):
      """Parses the recv.info file that is dumped by receiver.
  
      This file only contains the IP address of the receiver.
      """
  
      def __init__(self, rcvinf_file):
          f = open(rcvinf_file)
          self.ip = f.readlines()[0].strip()
          f.close()
  
  
  class ExpInfo(object):
      """Parses the exp.info file that is dumped by the orchestrator.
  
      This file contains a readable string representation of the experiment.
      """
  
      def __init__(self, expinf_file):
          f = open(expinf_file)
          self.__lines = f.readlines()
          f.close()
  
      def info(self):
          """Returns the lines in the exp.info file."""
          return self.__lines
  
      def fields(self):
          """Returns a dictionary of experiment parameters and their values."""
          field_dict = {}
          for l in self.__lines:
              p, v = l.strip().split('=', 1)
              field_dict[p] = v
          return field_dict
  