utilities/ovs-dpctl-top.in

   1 #! @PYTHON@
   2 #
   3 # Copyright (c) 2013 Nicira, Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at:
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 #
  18 # The approximate_size code was copied from
  19 # http://getpython3.com/diveintopython3/your-first-python-program.html#divingin
  20 # which is licensed under # "Dive Into Python 3," Copyright 2011 Mark Pilgrim,
  21 # used under a Creative Commons Attribution-Share-Alike license:
  22 # http://creativecommons.org/licenses/by-sa/3.0/
  23 #
  24 #
  25
  26 """Top like behavior for ovs-dpctl dump-flows output.
  27
  28 This program summarizes ovs-dpctl flow content by aggregating the number
  29 of packets, total bytes and occurrence of the following fields:
  30
  31   - Datapath in_port
  32
  33   - Ethernet type
  34
  35   - Source and destination MAC addresses
  36
  37   - IP protocol
  38
  39   - Source and destination IPv4 addresses
  40
  41   - Source and destination IPv6 addresses
  42
  43   - UDP and TCP destination port
  44
  45   - Tunnel source and destination addresses
  46
  47
  48 Output shows four values:
  49   - FIELDS: the flow fields for example in_port(1).
  50
  51   - PACKETS: the total number of packets containing the flow field.
  52
  53   - BYTES: the total number of bytes containing the flow field. If units are
  54   not present then values are in bytes.
  55
  56   - AVERAGE: the average packets size (BYTES/PACKET).
  57
  58   - COUNT: the number of lines in the dump-flow output contain the flow field.
  59
  60 Top Behavior
  61
  62 While in top mode, the default behavior, the following single character
  63 commands are supported:
  64
  65   a - toggles top in accumulate and live mode. Accumulate mode is described
  66   below.
  67
  68   s - toggles which column is used to sort content in decreasing order. A
  69   DESC title is placed over the column.
  70
  71   _ - a space indicating to collect dump-flow content again
  72
  73   h - halt output. Any character will restart sampling
  74
  75   f - cycle through flow fields. The initial field is in_port
  76
  77   q - q for quit.
  78
  79 Accumulate Mode
  80
  81 There are two supported modes: live and accumulate. The default is live.
  82 The parameter --accumulate  or the 'a' character in top mode enables the
  83 latter. In live mode, recent dump-flow content is presented.
  84 Where as accumulate mode keeps track of the prior historical
  85 information until the flow is reset not when the flow is purged. Reset
  86 flows are determined when the packet count for a flow has decreased from
  87 its previous sample. There is one caveat, eventually the system will
  88 run out of memory if, after the accumulate-decay period any flows that
  89 have not been refreshed are purged. The goal here is to free memory
  90 of flows that are not active. Statistics are not decremented. Their purpose
  91 is to reflect the overall history of the flow fields.
  92
  93
  94 Debugging Errors
  95
  96 Parsing errors are counted and displayed in the status line at the beginning
  97 of the output. Use the --verbose option with --script to see what output
  98  was not parsed, like this:
  99 $ ovs-dpctl dump-flows | ovs-dpctl-top --script --verbose
 100
 101 Error messages will identify content that failed to parse.
 102
 103
 104 Access Remote Hosts
 105
 106 The --host must follow the format user@hostname. This script simply calls
 107 'ssh user@Hostname' without checking for login credentials therefore public
 108 keys should be installed on the system identified by hostname, such as:
 109
 110 $ ssh-copy-id user@hostname
 111
 112 Consult ssh-copy-id man pages for more details.
 113
 114
 115 Expected usage
 116
 117 $ ovs-dpctl-top
 118
 119 or to run as a script:
 120 $ ovs-dpctl dump-flows > dump-flows.log
 121 $ ovs-dpctl-top --script --flow-file dump-flows.log
 122
 123 """
 124
 125 # pylint: disable-msg=C0103
 126 # pylint: disable-msg=C0302
 127 # pylint: disable-msg=R0902
 128 # pylint: disable-msg=R0903
 129 # pylint: disable-msg=R0904
 130 # pylint: disable-msg=R0912
 131 # pylint: disable-msg=R0913
 132 # pylint: disable-msg=R0914
 133
 134 import sys
 135 import os
 136 try:
 137     ##
 138     # Arg parse is not installed on older Python distributions.
 139     # ovs ships with a version in the directory mentioned below.
 140     import argparse
 141 except ImportError:
 142     sys.path.append(os.path.join("@pkgdatadir@", "python"))
 143     import argparse
 144 import logging
 145 import re
 146 import unittest
 147 import copy
 148 import curses
 149 import operator
 150 import subprocess
 151 import fcntl
 152 import struct
 153 import termios
 154 import datetime
 155 import threading
 156 import time
 157 import socket
 158
 159
 160 ##
 161 # The following two definitions provide the necessary netaddr functionality.
 162 # Python netaddr module is not part of the core installation. Packaging
 163 # netaddr was involved and seems inappropriate given that only two
 164 # methods where used.
 165 def ipv4_to_network(ip_str):
 166     """ Calculate the network given a ipv4/mask value.
 167     If a mask is not present simply return ip_str.
 168     """
 169     pack_length = '!HH'
 170     try:
 171         (ip, mask) = ip_str.split("/")
 172     except ValueError:
 173         # just an ip address no mask.
 174         return ip_str
 175
 176     ip_p = socket.inet_pton(socket.AF_INET, ip)
 177     ip_t = struct.unpack(pack_length, ip_p)
 178     mask_t = struct.unpack(pack_length, socket.inet_pton(socket.AF_INET, mask))
 179     network_n = [ii & jj for (ii, jj) in zip(ip_t, mask_t)]
 180
 181     return socket.inet_ntop(socket.AF_INET,
 182                             struct.pack('!HH', network_n[0], network_n[1]))
 183
 184
 185 def ipv6_to_network(ip_str):
 186     """ Calculate the network given a ipv6/mask value.
 187     If a mask is not present simply return ip_str.
 188     """
 189     pack_length = '!HHHHHHHH'
 190     try:
 191         (ip, mask) = ip_str.split("/")
 192     except ValueError:
 193         # just an ip address no mask.
 194         return ip_str
 195
 196     ip_p = socket.inet_pton(socket.AF_INET6, ip)
 197     ip_t = struct.unpack(pack_length, ip_p)
 198     mask_t = struct.unpack(pack_length,
 199                            socket.inet_pton(socket.AF_INET6, mask))
 200     network_n = [ii & jj for (ii, jj) in zip(ip_t, mask_t)]
 201
 202     return socket.inet_ntop(socket.AF_INET6,
 203                             struct.pack(pack_length,
 204                                         network_n[0], network_n[1],
 205                                         network_n[2], network_n[3],
 206                                         network_n[4], network_n[5],
 207                                         network_n[6], network_n[7]))
 208
 209
 210 ##
 211 # columns displayed
 212 ##
 213 class Columns:
 214     """ Holds column specific content.
 215     Titles needs to be less than 8 characters.
 216     """
 217     VALUE_WIDTH = 9
 218     FIELDS = "fields"
 219     PACKETS = "packets"
 220     COUNT = "count"
 221     BYTES = "bytes"
 222     AVERAGE = "average"
 223
 224     def __init__(self):
 225         pass
 226
 227     @staticmethod
 228     def assoc_list(obj):
 229         """ Return a associated list. """
 230         return [(Columns.FIELDS, repr(obj)),
 231                 (Columns.PACKETS, obj.packets),
 232                 (Columns.BYTES, obj.bytes),
 233                 (Columns.COUNT, obj.count),
 234                 (Columns.AVERAGE, obj.average),
 235                 ]
 236
 237
 238 def element_eth_get(field_type, element, stats_dict):
 239     """ Extract eth frame src and dst from a dump-flow element."""
 240     fmt = "%s(src=%s,dst=%s)"
 241
 242     element = fmt % (field_type, element["src"], element["dst"])
 243     return SumData(field_type, element, stats_dict["packets"],
 244                    stats_dict["bytes"], element)
 245
 246
 247 def element_ipv4_get(field_type, element, stats_dict):
 248     """ Extract src and dst from a dump-flow element."""
 249     fmt = "%s(src=%s,dst=%s)"
 250     element_show = fmt % (field_type, element["src"], element["dst"])
 251
 252     element_key = fmt % (field_type, ipv4_to_network(element["src"]),
 253                          ipv4_to_network(element["dst"]))
 254
 255     return SumData(field_type, element_show, stats_dict["packets"],
 256                        stats_dict["bytes"], element_key)
 257
 258
 259 def element_tunnel_get(field_type, element, stats_dict):
 260     """ Extract src and dst from a tunnel."""
 261     return element_ipv4_get(field_type, element, stats_dict)
 262
 263
 264 def element_ipv6_get(field_type, element, stats_dict):
 265     """ Extract src and dst from a dump-flow element."""
 266
 267     fmt = "%s(src=%s,dst=%s)"
 268     element_show = fmt % (field_type, element["src"], element["dst"])
 269
 270     element_key = fmt % (field_type, ipv6_to_network(element["src"]),
 271                          ipv6_to_network(element["dst"]))
 272
 273     return SumData(field_type, element_show, stats_dict["packets"],
 274                        stats_dict["bytes"], element_key)
 275
 276
 277 def element_dst_port_get(field_type, element, stats_dict):
 278     """ Extract src and dst from a dump-flow element."""
 279     element_key = "%s(dst=%s)" % (field_type, element["dst"])
 280     return SumData(field_type, element_key, stats_dict["packets"],
 281                    stats_dict["bytes"], element_key)
 282
 283
 284 def element_passthrough_get(field_type, element, stats_dict):
 285     """ Extract src and dst from a dump-flow element."""
 286     element_key = "%s(%s)" % (field_type, element)
 287     return SumData(field_type, element_key,
 288                    stats_dict["packets"], stats_dict["bytes"], element_key)
 289
 290
 291 # pylint: disable-msg=R0903
 292 class OutputFormat:
 293     """ Holds field_type and function to extract element value. """
 294     def __init__(self, field_type, generator):
 295         self.field_type = field_type
 296         self.generator = generator
 297
 298 ##
 299 # The order below is important. The initial flow field depends on whether
 300 # --script or top mode is used. In top mode, the expected behavior, in_port
 301 # flow fields are shown first. A future feature will allow users to
 302 # filter output by selecting a row. Filtering by in_port is a natural
 303 # filtering starting point.
 304 #
 305 # In script mode, all fields are shown. The expectation is that users could
 306 # filter output by piping through grep.
 307 #
 308 # In top mode, the default flow field is in_port. In --script mode,
 309 # the default flow field is all.
 310 #
 311 # All is added to the end of the OUTPUT_FORMAT list.
 312 ##
 313 OUTPUT_FORMAT = [
 314     OutputFormat("in_port", element_passthrough_get),
 315     OutputFormat("eth", element_eth_get),
 316     OutputFormat("eth_type", element_passthrough_get),
 317     OutputFormat("ipv4", element_ipv4_get),
 318     OutputFormat("ipv6", element_ipv6_get),
 319     OutputFormat("udp", element_dst_port_get),
 320     OutputFormat("tcp", element_dst_port_get),
 321     OutputFormat("tunnel", element_tunnel_get),
 322     ]
 323 ##
 324
 325
 326 ELEMENT_KEY = {
 327     "udp": "udp.dst",
 328     "tcp": "tcp.dst"
 329     }
 330
 331
 332 def top_input_get(args):
 333     """ Return subprocess stdout."""
 334     cmd = []
 335     if (args.host):
 336         cmd += ["ssh", args.host]
 337     cmd += ["ovs-dpctl", "dump-flows"]
 338
 339     return subprocess.Popen(cmd, stderr=subprocess.STDOUT,
 340                             stdout=subprocess.PIPE).stdout
 341
 342
 343 def args_get():
 344     """ read program parameters handle any necessary validation of input. """
 345
 346     parser = argparse.ArgumentParser(
 347                           formatter_class=argparse.RawDescriptionHelpFormatter,
 348                           description=__doc__)
 349     ##
 350     # None is a special value indicating to read flows from stdin.
 351     # This handles the case
 352     #   ovs-dpctl dump-flows | ovs-dpctl-flows.py
 353     parser.add_argument("-v", "--version", version="@VERSION@",
 354                         action="version", help="show version")
 355     parser.add_argument("-f", "--flow-file", dest="flowFiles", default=None,
 356                         action="append",
 357                         help="file containing flows from ovs-dpctl dump-flow")
 358     parser.add_argument("-V", "--verbose", dest="verbose",
 359                         default=logging.CRITICAL,
 360                         action="store_const", const=logging.DEBUG,
 361                         help="enable debug level verbosity")
 362     parser.add_argument("-s", "--script", dest="top", action="store_false",
 363                         help="Run from a script (no user interface)")
 364     parser.add_argument("--host", dest="host",
 365                         help="Specify a user@host for retrieving flows see"
 366                              "Accessing Remote Hosts for more information")
 367
 368     parser.add_argument("-a", "--accumulate", dest="accumulate",
 369                         action="store_true", default=False,
 370                         help="Accumulate dump-flow content")
 371     parser.add_argument("--accumulate-decay", dest="accumulateDecay",
 372                         default=5.0 * 60, type=float,
 373                         help="Decay old accumulated flows. "
 374                         "The default is 5 minutes. "
 375                         "A value of 0 disables decay.")
 376     parser.add_argument("-d", "--delay", dest="delay", type=int,
 377                         default=1000,
 378                         help="Delay in milliseconds to collect dump-flow "
 379                              "content (sample rate).")
 380
 381     args = parser.parse_args()
 382
 383     logging.basicConfig(level=args.verbose)
 384
 385     return args
 386
 387 ###
 388 # Code to parse a single line in dump-flow
 389 ###
 390 # key(values)
 391 FIELDS_CMPND = re.compile("([\w]+)\((.+)\)")
 392 # key:value
 393 FIELDS_CMPND_ELEMENT = re.compile("([\w:]+)=([/\.\w:]+)")
 394 FIELDS_ELEMENT = re.compile("([\w]+):([-\.\w]+)")
 395
 396
 397 def flow_line_iter(line):
 398     """ iterate over flow dump elements.
 399     return tuples of (true, element) or (false, remaining element)
 400     """
 401     # splits by , except for when in a (). Actions element was not
 402     # split properly but we don't need it.
 403     rc = []
 404
 405     element = ""
 406     paren_count = 0
 407
 408     for ch in line:
 409         if (ch == '('):
 410             paren_count += 1
 411         elif (ch == ')'):
 412             paren_count -= 1
 413
 414         if (ch == ' '):
 415             # ignore white space.
 416             continue
 417         elif ((ch == ',') and (paren_count == 0)):
 418             rc.append(element)
 419             element = ""
 420         else:
 421             element += ch
 422
 423     if (paren_count):
 424         raise ValueError(line)
 425     else:
 426         if (len(element) > 0):
 427             rc.append(element)
 428     return rc
 429
 430
 431 def flow_line_compound_parse(compound):
 432     """ Parse compound element
 433     for example
 434     src=00:50:56:b4:4e:f8,dst=33:33:00:01:00:03
 435     which is in
 436     eth(src=00:50:56:b4:4e:f8,dst=33:33:00:01:00:03)
 437     """
 438     result = {}
 439     for element in flow_line_iter(compound):
 440         match = FIELDS_CMPND_ELEMENT.search(element)
 441         if (match):
 442             key = match.group(1)
 443             value = match.group(2)
 444             result[key] = value
 445
 446         match = FIELDS_CMPND.search(element)
 447         if (match):
 448             key = match.group(1)
 449             value = match.group(2)
 450             result[key] = flow_line_compound_parse(value)
 451             continue
 452
 453     if (len(result.keys()) == 0):
 454         return compound
 455     return result
 456
 457
 458 def flow_line_split(line):
 459     """ Convert a flow dump line into ([fields], [stats], actions) tuple.
 460     Where fields and stats are lists.
 461     This function relies on a the following ovs-dpctl dump-flow
 462     output characteristics:
 463     1. The dumpe flow line consists of a list of frame fields, list of stats
 464        and action.
 465     2. list of frame fields, each stat and action field are delimited by ', '.
 466     3. That all other non stat field are not delimited by ', '.
 467
 468     """
 469
 470     results = re.split(', ', line)
 471
 472     (field, stats, action) = (results[0], results[1:-1], results[-1])
 473
 474     fields = flow_line_iter(field)
 475     return (fields, stats, action)
 476
 477
 478 def elements_to_dict(elements):
 479     """ Convert line to a hierarchy of dictionaries. """
 480     result = {}
 481     for element in elements:
 482         match = FIELDS_CMPND.search(element)
 483         if (match):
 484             key = match.group(1)
 485             value = match.group(2)
 486             result[key] = flow_line_compound_parse(value)
 487             continue
 488
 489         match = FIELDS_ELEMENT.search(element)
 490         if (match):
 491             key = match.group(1)
 492             value = match.group(2)
 493             result[key] = value
 494         else:
 495             raise ValueError("can't parse >%s<" % element)
 496     return result
 497
 498
 499 # pylint: disable-msg=R0903
 500 class SumData(object):
 501     """ Interface that all data going into SumDb must implement.
 502     Holds the flow field and its corresponding count, total packets,
 503     total bytes and calculates average.
 504
 505     __repr__ is used as key into SumData singleton.
 506     __str__ is used as human readable output.
 507     """
 508
 509     def __init__(self, field_type, field, packets, flow_bytes, key):
 510         # Count is the number of lines in the dump-flow log.
 511         self.field_type = field_type
 512         self.field = field
 513         self.count = 1
 514         self.packets = int(packets)
 515         self.bytes = int(flow_bytes)
 516         self.key = key
 517
 518     def decrement(self, decr_packets, decr_bytes, decr_count):
 519         """ Decrement content to calculate delta from previous flow sample."""
 520         self.packets -= decr_packets
 521         self.bytes -= decr_bytes
 522         self.count -= decr_count
 523
 524     def __iadd__(self, other):
 525         """ Add two objects. """
 526
 527         if (self.key != other.key):
 528             raise ValueError("adding two unrelated types")
 529
 530         self.count += other.count
 531         self.packets += other.packets
 532         self.bytes += other.bytes
 533         return self
 534
 535     def __isub__(self, other):
 536         """ Decrement two objects. """
 537
 538         if (self.key != other.key):
 539             raise ValueError("adding two unrelated types")
 540
 541         self.count -= other.count
 542         self.packets -= other.packets
 543         self.bytes -= other.bytes
 544         return self
 545
 546     def __getattr__(self, name):
 547         """ Handle average. """
 548         if (name == "average"):
 549             if (self.packets == 0):
 550                 return float(0.0)
 551             else:
 552                 return float(self.bytes) / float(self.packets)
 553         raise AttributeError(name)
 554
 555     def __str__(self):
 556         """ Used for debugging. """
 557         return "%s %s %s %s" % (self.field, self.count,
 558                                    self.packets, self.bytes)
 559
 560     def __repr__(self):
 561         """ Used as key in the FlowDB table. """
 562         return self.key
 563
 564
 565 def flow_aggregate(fields_dict, stats_dict):
 566     """ Search for content in a line.
 567     Passed the flow port of the dump-flows plus the current stats consisting
 568     of packets, bytes, etc
 569     """
 570     result = []
 571
 572     for output_format in OUTPUT_FORMAT:
 573         field = fields_dict.get(output_format.field_type, None)
 574         if (field):
 575             obj = output_format.generator(output_format.field_type,
 576                                           field, stats_dict)
 577             result.append(obj)
 578
 579     return result
 580
 581
 582 def flows_read(ihdl, flow_db):
 583     """ read flow content from ihdl and insert into flow_db. """
 584
 585     done = False
 586     while (not done):
 587         line = ihdl.readline()
 588         if (len(line) == 0):
 589             # end of input
 590             break
 591
 592         try:
 593             flow_db.flow_line_add(line)
 594         except ValueError, arg:
 595             logging.error(arg)
 596
 597     return flow_db
 598
 599
 600 def get_terminal_size():
 601     """
 602     return column width and height of the terminal
 603     """
 604     for fd_io in [0, 1, 2]:
 605         try:
 606             result = struct.unpack('hh',
 607                                    fcntl.ioctl(fd_io, termios.TIOCGWINSZ,
 608                                                '1234'))
 609         except IOError:
 610             result = None
 611             continue
 612
 613     if (result is None or result == (0, 0)):
 614         # Maybe we can't get the width. In that case assume (25, 80)
 615         result = (25, 80)
 616
 617     return result
 618
 619 ##
 620 # Content derived from:
 621 # http://getpython3.com/diveintopython3/your-first-python-program.html#divingin
 622 ##
 623 SUFFIXES = {1000: ['KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB'],
 624             1024: ['KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB', 'ZiB', 'YiB']}
 625
 626
 627 def approximate_size(size, a_kilobyte_is_1024_bytes=True):
 628     """Convert a file size to human-readable form.
 629
 630     Keyword arguments:
 631     size -- file size in bytes
 632     a_kilobyte_is_1024_bytes -- if True (default), use multiples of 1024
 633                                     if False, use multiples of 1000
 634
 635     Returns: string
 636
 637     """
 638     size = float(size)
 639     if size < 0:
 640         raise ValueError('number must be non-negative')
 641
 642     if (a_kilobyte_is_1024_bytes):
 643         multiple = 1024
 644     else:
 645         multiple = 1000
 646     for suffix in SUFFIXES[multiple]:
 647         size /= multiple
 648         if size < multiple:
 649             return "%.1f %s" % (size, suffix)
 650
 651     raise ValueError('number too large')
 652
 653
 654 ##
 655 # End copied content
 656 ##
 657 class ColMeta:
 658     """ Concepts about columns. """
 659     def __init__(self, sortable, width):
 660         self.sortable = sortable
 661         self.width = width
 662
 663
 664 class RowMeta:
 665     """ How to render rows. """
 666     def __init__(self, label, fmt):
 667         self.label = label
 668         self.fmt = fmt
 669
 670
 671 def fmt_packet(obj, width):
 672     """ Provide a string for packets that is appropriate for output."""
 673     return str(obj.packets).rjust(width)
 674
 675
 676 def fmt_count(obj, width):
 677     """ Provide a string for average that is appropriate for output."""
 678     return str(obj.count).rjust(width)
 679
 680
 681 def fmt_avg(obj, width):
 682     """ Provide a string for average that is appropriate for output."""
 683     return str(int(obj.average)).rjust(width)
 684
 685
 686 def fmt_field(obj, width):
 687     """ truncate really long flow and insert ellipses to help make it
 688     clear.
 689     """
 690
 691     ellipses = " ... "
 692     value = obj.field
 693     if (len(obj.field) > width):
 694         value = value[:(width - len(ellipses))] + ellipses
 695     return value.ljust(width)
 696
 697
 698 def fmt_bytes(obj, width):
 699     """ Provide a string for average that is appropriate for output."""
 700     if (len(str(obj.bytes)) <= width):
 701         value = str(obj.bytes)
 702     else:
 703         value = approximate_size(obj.bytes)
 704     return value.rjust(width)
 705
 706
 707 def title_center(value, width):
 708     """ Center a column title."""
 709     return value.upper().center(width)
 710
 711
 712 def title_rjust(value, width):
 713     """ Right justify a column title. """
 714     return value.upper().rjust(width)
 715
 716
 717 def column_picker(order, obj):
 718     """ return the column as specified by order. """
 719     if (order == 1):
 720         return obj.count
 721     elif (order == 2):
 722         return obj.packets
 723     elif (order == 3):
 724         return obj.bytes
 725     elif (order == 4):
 726         return obj.average
 727     else:
 728         raise ValueError("order outside of range %s" % order)
 729
 730
 731 class Render:
 732     """ Renders flow data.
 733
 734     The two FIELD_SELECT variables should be set to the actual field minus
 735     1. During construction, an internal method increments and initializes
 736     this object.
 737     """
 738     FLOW_FIELDS = [_field.field_type for _field in OUTPUT_FORMAT] + ["all"]
 739
 740     FIELD_SELECT_SCRIPT = 7
 741     FIELD_SELECT_TOP = -1
 742
 743     def __init__(self, console_width, field_select):
 744         """ Calculate column widths taking into account changes in format."""
 745
 746         self._start_time = datetime.datetime.now()
 747
 748         self._cols = [ColMeta(False, 0),
 749                       ColMeta(True, Columns.VALUE_WIDTH),
 750                       ColMeta(True, Columns.VALUE_WIDTH),
 751                       ColMeta(True, Columns.VALUE_WIDTH),
 752                       ColMeta(True, Columns.VALUE_WIDTH)]
 753         self._console_width = console_width
 754         self.console_width_set(console_width)
 755
 756         # Order in this array dictate the order of the columns.
 757         # The 0 width for the first entry is a place holder. This is
 758         # dynamically calculated. The first column is special. We need a
 759         # way to indicate which field are presented.
 760         self._descs = [RowMeta("", title_rjust),
 761                        RowMeta("", title_rjust),
 762                        RowMeta("", title_rjust),
 763                        RowMeta("", title_rjust),
 764                        RowMeta("", title_rjust)]
 765         self._column_sort_select = 0
 766         self.column_select_event()
 767
 768         self._titles = [
 769             RowMeta(Columns.FIELDS, title_center),
 770             RowMeta(Columns.COUNT, title_rjust),
 771             RowMeta(Columns.PACKETS, title_rjust),
 772             RowMeta(Columns.BYTES, title_rjust),
 773             RowMeta(Columns.AVERAGE, title_rjust)
 774         ]
 775
 776         self._datas = [
 777             RowMeta(None, fmt_field),
 778             RowMeta(None, fmt_count),
 779             RowMeta(None, fmt_packet),
 780             RowMeta(None, fmt_bytes),
 781             RowMeta(None, fmt_avg)
 782             ]
 783
 784         ##
 785         # _field_types hold which fields are displayed in the field
 786         # column, with the keyword all implying all fields.
 787         ##
 788         self._field_types = Render.FLOW_FIELDS
 789
 790         ##
 791         # The default is to show all field types.
 792         ##
 793         self._field_type_select = field_select
 794         self.field_type_toggle()
 795
 796     def _field_type_select_get(self):
 797         """ Return which field type to display. """
 798         return self._field_types[self._field_type_select]
 799
 800     def field_type_toggle(self):
 801         """ toggle which field types to show. """
 802         self._field_type_select += 1
 803         if (self._field_type_select >= len(self._field_types)):
 804             self._field_type_select = 0
 805         value = Columns.FIELDS + " (%s)" % self._field_type_select_get()
 806         self._titles[0].label = value
 807
 808     def column_select_event(self):
 809         """ Handles column select toggle. """
 810
 811         self._descs[self._column_sort_select].label = ""
 812         for _ in range(len(self._cols)):
 813             self._column_sort_select += 1
 814             if (self._column_sort_select >= len(self._cols)):
 815                 self._column_sort_select = 0
 816
 817             # Now look for the next sortable column
 818             if (self._cols[self._column_sort_select].sortable):
 819                 break
 820         self._descs[self._column_sort_select].label = "DESC"
 821
 822     def console_width_set(self, console_width):
 823         """ Adjust the output given the new console_width. """
 824         self._console_width = console_width
 825
 826         spaces = len(self._cols) - 1
 827         ##
 828         # Calculating column width can be tedious but important. The
 829         # flow field value can be long. The goal here is to dedicate
 830         # fixed column space for packets, bytes, average and counts. Give the
 831         # remaining space to the flow column. When numbers get large
 832         # transition output to output generated by approximate_size which
 833         # limits output to ###.# XiB in other words 9 characters.
 834         ##
 835         # At this point, we know the maximum length values. We may
 836         # truncate the flow column to get everything to fit.
 837         self._cols[0].width = 0
 838         values_max_length = sum([ii.width for ii in self._cols]) + spaces
 839         flow_max_length = console_width - values_max_length
 840         self._cols[0].width = flow_max_length
 841
 842     def format(self, flow_db):
 843         """ shows flows based on --script parameter."""
 844
 845         rc = []
 846         ##
 847         # Top output consists of
 848         # Title
 849         # Column title (2 rows)
 850         # data
 851         # statistics and status
 852
 853         ##
 854         # Title
 855         ##
 856         rc.append("Flow Summary".center(self._console_width))
 857
 858         stats = " Total: %(flow_total)s  errors: %(flow_errors)s " % \
 859                   flow_db.flow_stats_get()
 860         accumulate = flow_db.accumulate_get()
 861         if (accumulate):
 862             stats += "Accumulate: on "
 863         else:
 864             stats += "Accumulate: off "
 865
 866         duration = datetime.datetime.now() - self._start_time
 867         stats += "Duration: %s " % str(duration)
 868         rc.append(stats.ljust(self._console_width))
 869
 870         ##
 871         # 2 rows for columns.
 872         ##
 873         # Indicate which column is in descending order.
 874         rc.append(" ".join([ii.fmt(ii.label, col.width)
 875                             for (ii, col) in zip(self._descs, self._cols)]))
 876
 877         rc.append(" ".join([ii.fmt(ii.label, col.width)
 878                          for (ii, col) in zip(self._titles, self._cols)]))
 879
 880         ##
 881         # Data.
 882         ##
 883         for dd in flow_db.field_values_in_order(self._field_type_select_get(),
 884                                                 self._column_sort_select):
 885             rc.append(" ".join([ii.fmt(dd, col.width)
 886                                 for (ii, col) in zip(self._datas,
 887                                                      self._cols)]))
 888
 889         return rc
 890
 891
 892 def curses_screen_begin():
 893     """ begin curses screen control. """
 894     stdscr = curses.initscr()
 895     curses.cbreak()
 896     curses.noecho()
 897     stdscr.keypad(1)
 898     return stdscr
 899
 900
 901 def curses_screen_end(stdscr):
 902     """ end curses screen control. """
 903     curses.nocbreak()
 904     stdscr.keypad(0)
 905     curses.echo()
 906     curses.endwin()
 907
 908
 909 class FlowDB:
 910     """ Implements live vs accumulate mode.
 911
 912     Flows are stored as key value pairs. The key consists of the content
 913     prior to stat fields. The value portion consists of stats in a dictionary
 914     form.
 915
 916     @ \todo future add filtering here.
 917     """
 918     def __init__(self, accumulate):
 919         self._accumulate = accumulate
 920         self._error_count = 0
 921         # Values are (stats, last update time.)
 922         # The last update time is used for aging.
 923         self._flow_lock = threading.Lock()
 924         # This dictionary holds individual flows.
 925         self._flows = {}
 926         # This dictionary holds aggregate of flow fields.
 927         self._fields = {}
 928
 929     def accumulate_get(self):
 930         """ Return the current accumulate state. """
 931         return self._accumulate
 932
 933     def accumulate_toggle(self):
 934         """ toggle accumulate flow behavior. """
 935         self._accumulate = not self._accumulate
 936
 937     def begin(self):
 938         """ Indicate the beginning of processing flow content.
 939         if accumulate is false clear current set of flows. """
 940
 941         if (not self._accumulate):
 942             self._flow_lock.acquire()
 943             try:
 944                 self._flows.clear()
 945             finally:
 946                 self._flow_lock.release()
 947             self._fields.clear()
 948
 949     def flow_line_add(self, line):
 950         """ Split a line from a ovs-dpctl dump-flow into key and stats.
 951         The order of the content in the flow should be:
 952         - flow content
 953         - stats for the flow
 954         - actions
 955
 956         This method also assumes that the dump flow output does not
 957         change order of fields of the same flow.
 958         """
 959
 960         line = line.rstrip("\n")
 961         (fields, stats, _) = flow_line_split(line)
 962
 963         try:
 964             fields_dict = elements_to_dict(fields)
 965
 966             if (len(fields_dict) == 0):
 967                 raise ValueError("flow fields are missing %s", line)
 968
 969             stats_dict = elements_to_dict(stats)
 970             if (len(stats_dict) == 0):
 971                 raise ValueError("statistics are missing %s.", line)
 972
 973             ##
 974             # In accumulate mode, the Flow database can reach 10,000's of
 975             # persistent flows. The interaction of the script with this many
 976             # flows is too slow. Instead, delta are sent to the flow_db
 977             # database allow incremental changes to be done in O(m) time
 978             # where m is the current flow list, instead of iterating over
 979             # all flows in O(n) time where n is the entire history of flows.
 980             key = ",".join(fields)
 981
 982             self._flow_lock.acquire()
 983             try:
 984                 (stats_old_dict, _) = self._flows.get(key, (None, None))
 985             finally:
 986                 self._flow_lock.release()
 987
 988             self.flow_event(fields_dict, stats_old_dict, stats_dict)
 989
 990         except ValueError, arg:
 991             logging.error(arg)
 992             self._error_count += 1
 993             raise
 994
 995         self._flow_lock.acquire()
 996         try:
 997             self._flows[key] = (stats_dict, datetime.datetime.now())
 998         finally:
 999             self._flow_lock.release()
1000
1001     def decay(self, decayTimeInSeconds):
1002         """ Decay content. """
1003         now = datetime.datetime.now()
1004         for (key, value) in self._flows.items():
1005             (stats_dict, updateTime) = value
1006             delta = now - updateTime
1007
1008             if (delta.seconds > decayTimeInSeconds):
1009                 self._flow_lock.acquire()
1010                 try:
1011                     del self._flows[key]
1012
1013                     fields_dict = elements_to_dict(flow_line_iter(key))
1014                     matches = flow_aggregate(fields_dict, stats_dict)
1015                     for match in matches:
1016                         self.field_dec(match)
1017
1018                 finally:
1019                     self._flow_lock.release()
1020
1021     def flow_stats_get(self):
1022         """ Return statistics in a form of a dictionary. """
1023         rc = None
1024         self._flow_lock.acquire()
1025         try:
1026             rc = {"flow_total": len(self._flows),
1027                   "flow_errors": self._error_count}
1028         finally:
1029             self._flow_lock.release()
1030         return rc
1031
1032     def field_types_get(self):
1033         """ Return the set of types stored in the singleton. """
1034         types = set((ii.field_type for ii in self._fields.values()))
1035         return types
1036
1037     def field_add(self, data):
1038         """ Collect dump-flow data to sum number of times item appears. """
1039         current = self._fields.get(repr(data), None)
1040         if (current is None):
1041             current = copy.copy(data)
1042         else:
1043             current += data
1044         self._fields[repr(current)] = current
1045
1046     def field_dec(self, data):
1047         """ Collect dump-flow data to sum number of times item appears. """
1048         current = self._fields.get(repr(data), None)
1049         if (current is None):
1050             raise ValueError("decrementing field missing %s" % repr(data))
1051
1052         current -= data
1053         self._fields[repr(current)] = current
1054         if (current.count == 0):
1055             del self._fields[repr(current)]
1056
1057     def field_values_in_order(self, field_type_select, column_order):
1058         """ Return a list of items in order maximum first. """
1059         values = self._fields.values()
1060         if (field_type_select != "all"):
1061             # If a field type other than "all" then reduce the list.
1062             values = [ii for ii in values
1063                       if (ii.field_type == field_type_select)]
1064         values = [(column_picker(column_order, ii), ii) for ii in values]
1065         values.sort(key=operator.itemgetter(0))
1066         values.reverse()
1067         values = [ii[1] for ii in values]
1068         return values
1069
1070     def flow_event(self, fields_dict, stats_old_dict, stats_new_dict):
1071         """ Receives new flow information. """
1072
1073         # In order to avoid processing every flow at every sample
1074         # period, changes in flow packet count is used to determine the
1075         # delta in the flow statistics. This delta is used in the call
1076         # to self.decrement prior to self.field_add
1077
1078         if (stats_old_dict is None):
1079             # This is a new flow
1080             matches = flow_aggregate(fields_dict, stats_new_dict)
1081             for match in matches:
1082                 self.field_add(match)
1083         else:
1084             old_packets = int(stats_old_dict.get("packets", 0))
1085             new_packets = int(stats_new_dict.get("packets", 0))
1086             if (old_packets == new_packets):
1087                 # ignore. same data.
1088                 pass
1089             else:
1090                 old_bytes = stats_old_dict.get("bytes", 0)
1091                 # old_packets != new_packets
1092                 # if old_packets > new_packets then we end up decrementing
1093                 # packets and bytes.
1094                 matches = flow_aggregate(fields_dict, stats_new_dict)
1095                 for match in matches:
1096                     match.decrement(int(old_packets), int(old_bytes), 1)
1097                     self.field_add(match)
1098
1099
1100 class DecayThread(threading.Thread):
1101     """ Periodically call flow database to see if any flows are old. """
1102     def __init__(self, flow_db, interval):
1103         """ Start decay thread. """
1104         threading.Thread.__init__(self)
1105
1106         self._interval = max(1, interval)
1107         self._min_interval = min(1, interval / 10)
1108         self._flow_db = flow_db
1109         self._event = threading.Event()
1110         self._running = True
1111
1112         self.daemon = True
1113
1114     def run(self):
1115         """ Worker thread which handles decaying accumulated flows. """
1116
1117         while(self._running):
1118             self._event.wait(self._min_interval)
1119             if (self._running):
1120                 self._flow_db.decay(self._interval)
1121
1122     def stop(self):
1123         """ Stop thread. """
1124         self._running = False
1125         self._event.set()
1126         ##
1127         # Give the calling thread time to terminate but not too long.
1128         # this thread is a daemon so the application will terminate if
1129         # we timeout during the join. This is just a cleaner way to
1130         # release resources.
1131         self.join(2.0)
1132
1133
1134 def flow_top_command(stdscr, render, flow_db):
1135     """ Handle input while in top mode. """
1136     ch = stdscr.getch()
1137     ##
1138     # Any character will restart sampling.
1139     if (ch == ord('h')):
1140         # halt output.
1141         ch = stdscr.getch()
1142         while (ch == -1):
1143             ch = stdscr.getch()
1144
1145     if (ch == ord('s')):
1146         # toggle which column sorts data in descending order.
1147         render.column_select_event()
1148     elif (ch == ord('a')):
1149         flow_db.accumulate_toggle()
1150     elif (ch == ord('f')):
1151         render.field_type_toggle()
1152     elif (ch == ord(' ')):
1153         # resample
1154         pass
1155
1156     return ch
1157
1158
1159 def decay_timer_start(flow_db, accumulateDecay):
1160     """ If accumulateDecay greater than zero then start timer. """
1161     if (accumulateDecay > 0):
1162         decay_timer = DecayThread(flow_db, accumulateDecay)
1163         decay_timer.start()
1164         return decay_timer
1165     else:
1166         return None
1167
1168
1169 def flows_top(args):
1170     """ handles top like behavior when --script is not specified. """
1171
1172     flow_db = FlowDB(args.accumulate)
1173     render = Render(0, Render.FIELD_SELECT_TOP)
1174
1175     decay_timer = decay_timer_start(flow_db, args.accumulateDecay)
1176     lines = []
1177
1178     try:
1179         stdscr = curses_screen_begin()
1180         try:
1181             ch = 'X'
1182             #stdscr.nodelay(1)
1183             stdscr.timeout(args.delay)
1184
1185             while (ch != ord('q')):
1186                 flow_db.begin()
1187
1188                 try:
1189                     ihdl = top_input_get(args)
1190                     try:
1191                         flows_read(ihdl, flow_db)
1192                     finally:
1193                         ihdl.close()
1194                 except OSError, arg:
1195                     logging.critical(arg)
1196                     break
1197
1198                 (console_height, console_width) = stdscr.getmaxyx()
1199                 render.console_width_set(console_width)
1200
1201                 output_height = console_height - 1
1202                 line_count = range(output_height)
1203                 line_output = render.format(flow_db)
1204                 lines = zip(line_count, line_output[:output_height])
1205
1206                 stdscr.erase()
1207                 for (count, line) in lines:
1208                     stdscr.addstr(count, 0, line[:console_width])
1209                 stdscr.refresh()
1210
1211                 ch = flow_top_command(stdscr, render, flow_db)
1212
1213         finally:
1214             curses_screen_end(stdscr)
1215     except KeyboardInterrupt:
1216         pass
1217     if (decay_timer):
1218         decay_timer.stop()
1219
1220     # repeat output
1221     for (count, line) in lines:
1222         print line
1223
1224
1225 def flows_script(args):
1226     """ handles --script option. """
1227
1228     flow_db = FlowDB(args.accumulate)
1229     flow_db.begin()
1230
1231     if (args.flowFiles is None):
1232         logging.info("reading flows from stdin")
1233         ihdl = os.fdopen(sys.stdin.fileno(), 'r', 0)
1234         try:
1235             flow_db = flows_read(ihdl, flow_db)
1236         finally:
1237             ihdl.close()
1238     else:
1239         for flowFile in args.flowFiles:
1240             logging.info("reading flows from %s", flowFile)
1241             ihdl = open(flowFile, "r")
1242             try:
1243                 flow_db = flows_read(ihdl, flow_db)
1244             finally:
1245                 ihdl.close()
1246
1247     (_, console_width) = get_terminal_size()
1248     render = Render(console_width, Render.FIELD_SELECT_SCRIPT)
1249
1250     for line in render.format(flow_db):
1251         print line
1252
1253
1254 def main():
1255     """ Return 0 on success or 1 on failure.
1256
1257     Algorithm
1258     There are four stages to the process ovs-dpctl dump-flow content.
1259     1. Retrieve current input
1260     2. store in FlowDB and maintain history
1261     3. Iterate over FlowDB and aggregating stats for each flow field
1262     4. present data.
1263
1264     Retrieving current input is currently trivial, the ovs-dpctl dump-flow
1265     is called. Future version will have more elaborate means for collecting
1266     dump-flow content. FlowDB returns all data as in the form of a hierarchical
1267     dictionary. Input will vary.
1268
1269     In the case of accumulate mode, flows are not purged from the FlowDB
1270     manager. Instead at the very least, merely the latest statistics are
1271     kept. In the case, of live output the FlowDB is purged prior to sampling
1272     data.
1273
1274     Aggregating results requires identify flow fields to aggregate out
1275     of the flow and summing stats.
1276
1277     """
1278     args = args_get()
1279
1280     try:
1281         if (args.top):
1282             flows_top(args)
1283         else:
1284             flows_script(args)
1285     except KeyboardInterrupt:
1286         return 1
1287     return 0
1288
1289 if __name__ == '__main__':
1290     sys.exit(main())
1291 elif __name__ == 'ovs-dpctl-top':
1292     # pylint: disable-msg=R0915
1293
1294     ##
1295     # Test case beyond this point.
1296     # pylint: disable-msg=R0904
1297     class TestsuiteFlowParse(unittest.TestCase):
1298         """
1299         parse flow into hierarchy of dictionaries.
1300         """
1301         def test_flow_parse(self):
1302             """ test_flow_parse. """
1303             line = "in_port(4),eth(src=00:50:56:b4:4e:f8,"\
1304                    "dst=33:33:00:01:00:03),eth_type(0x86dd),"\
1305                    "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"\
1306                    "label=0,proto=17,tclass=0,hlimit=1,frag=no),"\
1307                    "udp(src=61252,dst=5355), packets:1, bytes:92, "\
1308                    "used:0.703s, actions:3,8,11,14,17,20,23,26,29,32,35,"\
1309                    "38,41,44,47,50,53,56,59,62,65"
1310
1311             (fields, stats, _) = flow_line_split(line)
1312             flow_dict = elements_to_dict(fields + stats)
1313             self.assertEqual(flow_dict["eth"]["src"], "00:50:56:b4:4e:f8")
1314             self.assertEqual(flow_dict["eth"]["dst"], "33:33:00:01:00:03")
1315             self.assertEqual(flow_dict["ipv6"]["src"],
1316                              "fe80::55bf:fe42:bc96:2812")
1317             self.assertEqual(flow_dict["ipv6"]["dst"], "ff02::1:3")
1318             self.assertEqual(flow_dict["packets"], "1")
1319             self.assertEqual(flow_dict["bytes"], "92")
1320
1321             line = "in_port(4),eth(src=00:50:56:b4:4e:f8,"\
1322                    "dst=33:33:00:01:00:03),eth_type(0x86dd),"\
1323                    "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"\
1324                    "label=0,proto=17,tclass=0,hlimit=1,frag=no),"\
1325                    "udp(src=61252,dst=5355), packets:1, bytes:92, "\
1326                    "used:-0.703s, actions:3,8,11,14,17,20,23,26,29,32,35,"\
1327                    "38,41,44,47,50,53,56,59,62,65"
1328
1329             (fields, stats, _) = flow_line_split(line)
1330             flow_dict = elements_to_dict(fields + stats)
1331             self.assertEqual(flow_dict["used"], "-0.703s")
1332             self.assertEqual(flow_dict["packets"], "1")
1333             self.assertEqual(flow_dict["bytes"], "92")
1334
1335         def test_flow_sum(self):
1336             """ test_flow_sum. """
1337             line = "in_port(4),eth(src=00:50:56:b4:4e:f8,"\
1338                    "dst=33:33:00:01:00:03),eth_type(0x86dd),"\
1339                    "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"\
1340                    "label=0,proto=17,tclass=0,hlimit=1,frag=no),"\
1341                    "udp(src=61252,dst=5355), packets:2, bytes:92, "\
1342                    "used:0.703s, actions:3,8,11,14,17,20,23,26,29,32,35,"\
1343                    "38,41,44,47,50,53,56,59,62,65"
1344
1345             (fields, stats, _) = flow_line_split(line)
1346             stats_dict = elements_to_dict(stats)
1347             fields_dict = elements_to_dict(fields)
1348             ##
1349             # Test simple case of one line.
1350             flow_db = FlowDB(False)
1351             matches = flow_aggregate(fields_dict, stats_dict)
1352             for match in matches:
1353                 flow_db.field_add(match)
1354
1355             flow_types = flow_db.field_types_get()
1356             expected_flow_types = ["eth", "eth_type", "udp", "in_port", "ipv6"]
1357             self.assert_(len(flow_types) == len(expected_flow_types))
1358             for flow_type in flow_types:
1359                 self.assertTrue(flow_type in expected_flow_types)
1360
1361             for flow_type in flow_types:
1362                 sum_value = flow_db.field_values_in_order("all", 1)
1363                 self.assert_(len(sum_value) == 5)
1364                 self.assert_(sum_value[0].packets == 2)
1365                 self.assert_(sum_value[0].count == 1)
1366                 self.assert_(sum_value[0].bytes == 92)
1367
1368             ##
1369             # Add line again just to see counts go up.
1370             matches = flow_aggregate(fields_dict, stats_dict)
1371             for match in matches:
1372                 flow_db.field_add(match)
1373
1374             flow_types = flow_db.field_types_get()
1375             self.assert_(len(flow_types) == len(expected_flow_types))
1376             for flow_type in flow_types:
1377                 self.assertTrue(flow_type in expected_flow_types)
1378
1379             for flow_type in flow_types:
1380                 sum_value = flow_db.field_values_in_order("all", 1)
1381                 self.assert_(len(sum_value) == 5)
1382                 self.assert_(sum_value[0].packets == 4)
1383                 self.assert_(sum_value[0].count == 2)
1384                 self.assert_(sum_value[0].bytes == 2 * 92)
1385
1386         def test_assoc_list(self):
1387             """ test_assoc_list. """
1388             line = "in_port(4),eth(src=00:50:56:b4:4e:f8,"\
1389                    "dst=33:33:00:01:00:03),eth_type(0x86dd),"\
1390                    "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"\
1391                    "label=0,proto=17,tclass=0,hlimit=1,frag=no),"\
1392                    "udp(src=61252,dst=5355), packets:2, bytes:92, "\
1393                    "used:0.703s, actions:3,8,11,14,17,20,23,26,29,32,35,"\
1394                    "38,41,44,47,50,53,56,59,62,65"
1395
1396             valid_flows = [
1397                 'eth_type(0x86dd)',
1398                 'udp(dst=5355)',
1399                 'in_port(4)',
1400                 'ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3)',
1401                 'eth(src=00:50:56:b4:4e:f8,dst=33:33:00:01:00:03)'
1402                 ]
1403
1404             (fields, stats, _) = flow_line_split(line)
1405             stats_dict = elements_to_dict(stats)
1406             fields_dict = elements_to_dict(fields)
1407
1408             ##
1409             # Test simple case of one line.
1410             flow_db = FlowDB(False)
1411             matches = flow_aggregate(fields_dict, stats_dict)
1412             for match in matches:
1413                 flow_db.field_add(match)
1414
1415             for sum_value in flow_db.field_values_in_order("all", 1):
1416                 assoc_list = Columns.assoc_list(sum_value)
1417                 for item in assoc_list:
1418                     if (item[0] == "fields"):
1419                         self.assertTrue(item[1] in valid_flows)
1420                     elif (item[0] == "packets"):
1421                         self.assertTrue(item[1] == 2)
1422                     elif (item[0] == "count"):
1423                         self.assertTrue(item[1] == 1)
1424                     elif (item[0] == "average"):
1425                         self.assertTrue(item[1] == 46.0)
1426                     elif (item[0] == "bytes"):
1427                         self.assertTrue(item[1] == 92)
1428                     else:
1429                         raise ValueError("unknown %s", item[0])
1430
1431         def test_human_format(self):
1432             """ test_assoc_list. """
1433
1434             self.assertEqual(approximate_size(0.0), "0.0 KiB")
1435             self.assertEqual(approximate_size(1024), "1.0 KiB")
1436             self.assertEqual(approximate_size(1024 * 1024), "1.0 MiB")
1437             self.assertEqual(approximate_size((1024 * 1024) + 100000),
1438                              "1.1 MiB")
1439             value = (1024 * 1024 * 1024) + 100000000
1440             self.assertEqual(approximate_size(value), "1.1 GiB")
1441
1442         def test_flow_line_split(self):
1443             """ Splitting a flow line is not trivial.
1444             There is no clear delimiter. Comma is used liberally."""
1445             expected_fields = ["in_port(4)",
1446                             "eth(src=00:50:56:b4:4e:f8,dst=33:33:00:01:00:03)",
1447                             "eth_type(0x86dd)",
1448                            "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"
1449                            "label=0,proto=17,tclass=0,hlimit=1,frag=no)",
1450                            "udp(src=61252,dst=5355)"]
1451             expected_stats = ["packets:2", "bytes:92", "used:0.703s"]
1452             expected_actions = "actions:3,8,11,14,17,20,23,26,29,32,35," \
1453                                "38,41,44,47,50,53,56,59,62,65"
1454
1455             line = "in_port(4),eth(src=00:50:56:b4:4e:f8,"\
1456                    "dst=33:33:00:01:00:03),eth_type(0x86dd),"\
1457                    "ipv6(src=fe80::55bf:fe42:bc96:2812,dst=ff02::1:3,"\
1458                    "label=0,proto=17,tclass=0,hlimit=1,frag=no),"\
1459                    "udp(src=61252,dst=5355), packets:2, bytes:92, "\
1460                    "used:0.703s, actions:3,8,11,14,17,20,23,26,29,32,35,"\
1461                    "38,41,44,47,50,53,56,59,62,65"
1462
1463             (fields, stats, actions) = flow_line_split(line)
1464
1465             self.assertEqual(fields, expected_fields)
1466             self.assertEqual(stats, expected_stats)
1467             self.assertEqual(actions, expected_actions)
1468
1469         def test_accumulate_decay(self):
1470             """ test_accumulate_decay: test accumulated decay. """
1471             lines = ["in_port(1),eth(src=00:50:56:4f:dc:3b,"
1472                      "dst=ff:ff:ff:ff:ff:ff),"
1473                      "eth_type(0x0806),arp(sip=10.24.105.107/255.255.255.255,"
1474                      "tip=10.24.104.230/255.255.255.255,op=1/0xff,"
1475                      "sha=00:50:56:4f:dc:3b/00:00:00:00:00:00,"
1476                      "tha=00:00:00:00:00:00/00:00:00:00:00:00), "
1477                      "packets:1, bytes:120, used:0.004s, actions:1"]
1478
1479             flow_db = FlowDB(True)
1480             flow_db.begin()
1481             flow_db.flow_line_add(lines[0])
1482
1483             # Make sure we decay
1484             time.sleep(4)
1485             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 1)
1486             flow_db.decay(1)
1487             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 0)
1488
1489             flow_db.flow_line_add(lines[0])
1490             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 1)
1491             flow_db.decay(30)
1492             # Should not be deleted.
1493             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 1)
1494
1495             flow_db.flow_line_add(lines[0])
1496             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 1)
1497             timer = decay_timer_start(flow_db, 2)
1498             time.sleep(10)
1499             self.assertEqual(flow_db.flow_stats_get()["flow_total"], 0)
1500             timer.stop()
1501
1502         def test_accumulate(self):
1503             """ test_accumulate test that FlowDB supports accumulate. """
1504
1505             lines = ["in_port(1),eth(src=00:50:56:4f:dc:3b,"
1506                      "dst=ff:ff:ff:ff:ff:ff),"
1507                      "eth_type(0x0806),arp(sip=10.24.105.107/255.255.255.255,"
1508                      "tip=10.24.104.230/255.255.255.255,op=1/0xff,"
1509                      "sha=00:50:56:4f:dc:3b/00:00:00:00:00:00,"
1510                      "tha=00:00:00:00:00:00/00:00:00:00:00:00), "
1511                      "packets:1, bytes:120, used:0.004s, actions:1",
1512                      "in_port(2),"
1513                      "eth(src=68:ef:bd:25:ef:c0,dst=33:33:00:00:00:66),"
1514                      "eth_type(0x86dd),ipv6(src=fe80::6aef:bdff:fe25:efc0/::,"
1515                      "dst=ff02::66/::,label=0/0,proto=17/0xff,tclass=0xe0/0,"
1516                      "hlimit=255/0,frag=no/0),udp(src=2029,dst=2029), "
1517                      "packets:2, bytes:5026, used:0.348s, actions:1",
1518                      "in_port(1),eth(src=ee:ee:ee:ee:ee:ee,"
1519                      "dst=ff:ff:ff:ff:ff:ff),"
1520                      "eth_type(0x0806),arp(sip=10.24.105.107/255.255.255.255,"
1521                      "tip=10.24.104.230/255.255.255.255,op=1/0xff,"
1522                      "sha=00:50:56:4f:dc:3b/00:00:00:00:00:00,"
1523                      "tha=00:00:00:00:00:00/00:00:00:00:00:00), packets:2, "
1524                      "bytes:240, used:0.004s, actions:1"]
1525
1526             lines = [
1527                 "in_port(1),eth_type(0x0806), packets:1, bytes:120, actions:1",
1528                 "in_port(2),eth_type(0x0806), packets:2, bytes:126, actions:1",
1529                 "in_port(1),eth_type(0x0806), packets:2, bytes:240, actions:1",
1530                 "in_port(1),eth_type(0x0800), packets:1, bytes:120, actions:1",
1531                 "in_port(1),eth_type(0x0800), packets:2, bytes:240, actions:1",
1532                 "in_port(1),eth_type(0x0806), packets:1, bytes:120, actions:1",
1533                 ]
1534
1535             # Turn on accumulate.
1536             flow_db = FlowDB(True)
1537             flow_db.begin()
1538
1539             flow_db.flow_line_add(lines[0])
1540
1541             # Test one flow exist.
1542             sum_values = flow_db.field_values_in_order("all", 1)
1543             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1544             self.assertEqual(len(in_ports), 1)
1545             self.assertEqual(in_ports[0].packets, 1)
1546             self.assertEqual(in_ports[0].bytes, 120)
1547             self.assertEqual(in_ports[0].count, 1)
1548
1549             # simulate another sample
1550             # Test two different flows exist.
1551             flow_db.begin()
1552             flow_db.flow_line_add(lines[1])
1553             sum_values = flow_db.field_values_in_order("all", 1)
1554             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1555             self.assertEqual(len(in_ports), 1)
1556             self.assertEqual(in_ports[0].packets, 1)
1557             self.assertEqual(in_ports[0].bytes, 120)
1558             self.assertEqual(in_ports[0].count, 1)
1559
1560             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(2)")]
1561             self.assertEqual(len(in_ports), 1)
1562             self.assertEqual(in_ports[0].packets, 2)
1563             self.assertEqual(in_ports[0].bytes, 126)
1564             self.assertEqual(in_ports[0].count, 1)
1565
1566             # Test first flow increments packets.
1567             flow_db.begin()
1568             flow_db.flow_line_add(lines[2])
1569             sum_values = flow_db.field_values_in_order("all", 1)
1570             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1571             self.assertEqual(len(in_ports), 1)
1572             self.assertEqual(in_ports[0].packets, 2)
1573             self.assertEqual(in_ports[0].bytes, 240)
1574             self.assertEqual(in_ports[0].count, 1)
1575
1576             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(2)")]
1577             self.assertEqual(len(in_ports), 1)
1578             self.assertEqual(in_ports[0].packets, 2)
1579             self.assertEqual(in_ports[0].bytes, 126)
1580             self.assertEqual(in_ports[0].count, 1)
1581
1582             # Test third flow but with the same in_port(1) as the first flow.
1583             flow_db.begin()
1584             flow_db.flow_line_add(lines[3])
1585             sum_values = flow_db.field_values_in_order("all", 1)
1586             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1587             self.assertEqual(len(in_ports), 1)
1588             self.assertEqual(in_ports[0].packets, 3)
1589             self.assertEqual(in_ports[0].bytes, 360)
1590             self.assertEqual(in_ports[0].count, 2)
1591
1592             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(2)")]
1593             self.assertEqual(len(in_ports), 1)
1594             self.assertEqual(in_ports[0].packets, 2)
1595             self.assertEqual(in_ports[0].bytes, 126)
1596             self.assertEqual(in_ports[0].count, 1)
1597
1598             # Third flow has changes.
1599             flow_db.begin()
1600             flow_db.flow_line_add(lines[4])
1601             sum_values = flow_db.field_values_in_order("all", 1)
1602             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1603             self.assertEqual(len(in_ports), 1)
1604             self.assertEqual(in_ports[0].packets, 4)
1605             self.assertEqual(in_ports[0].bytes, 480)
1606             self.assertEqual(in_ports[0].count, 2)
1607
1608             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(2)")]
1609             self.assertEqual(len(in_ports), 1)
1610             self.assertEqual(in_ports[0].packets, 2)
1611             self.assertEqual(in_ports[0].bytes, 126)
1612             self.assertEqual(in_ports[0].count, 1)
1613
1614             # First flow reset.
1615             flow_db.begin()
1616             flow_db.flow_line_add(lines[5])
1617             sum_values = flow_db.field_values_in_order("all", 1)
1618             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1619             self.assertEqual(len(in_ports), 1)
1620             self.assertEqual(in_ports[0].packets, 3)
1621             self.assertEqual(in_ports[0].bytes, 360)
1622             self.assertEqual(in_ports[0].count, 2)
1623
1624             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(2)")]
1625             self.assertEqual(len(in_ports), 1)
1626             self.assertEqual(in_ports[0].packets, 2)
1627             self.assertEqual(in_ports[0].bytes, 126)
1628             self.assertEqual(in_ports[0].count, 1)
1629
1630         def test_parse_character_errors(self):
1631             """ test_parsing errors.
1632             The flow parses is purposely loose. Its not designed to validate
1633             input. Merely pull out what it can but there are situations
1634             that a parse error can be detected.
1635             """
1636
1637             lines = ["complete garbage",
1638                      "in_port(2),eth(src=68:ef:bd:25:ef:c0,"
1639                      "dst=33:33:00:00:00:66),"
1640                      "eth_type(0x86dd),ipv6(src=fe80::6aef:bdff:fe25:efc0/::,"
1641                      "dst=ff02::66/::,label=0/0,proto=17/0xff,tclass=0xe0/0,"
1642                      "hlimit=255/0,frag=no/0),udp(src=2029,dst=2029),"
1643                      "packets:2,bytes:5026,actions:1"]
1644
1645             flow_db = FlowDB(False)
1646             flow_db.begin()
1647             for line in lines:
1648                 try:
1649                     flow_db.flow_line_add(line)
1650                 except ValueError:
1651                     # We want an exception. That is how we know we have
1652                     # correctly found a simple parsing error. We are not
1653                     # looking to validate flow output just catch simple issues.
1654                     continue
1655                 self.assertTrue(False)
1656
1657         def test_tunnel_parsing(self):
1658             """ test_tunnel_parsing test parse flows with tunnel. """
1659             lines = [
1660                 "tunnel(tun_id=0x0,src=192.168.1.1,dst=192.168.1.10,"
1661                 "tos=0x0,ttl=64,flags(key)),in_port(1),"
1662                 "eth(src=9e:40:f5:ef:ec:ee,dst=01:23:20:00:00:30),"
1663                 "eth_type(0x8902), packets:6, bytes:534, used:0.128s, "
1664                 "actions:userspace(pid=4294962691,slow_path(cfm))"
1665                 ]
1666             flow_db = FlowDB(False)
1667             flow_db.begin()
1668             flow_db.flow_line_add(lines[0])
1669             sum_values = flow_db.field_values_in_order("all", 1)
1670             in_ports = [ii for ii in sum_values if (repr(ii) == "in_port(1)")]
1671             self.assertEqual(len(in_ports), 1)
1672             self.assertEqual(in_ports[0].packets, 6)
1673             self.assertEqual(in_ports[0].bytes, 534)
1674             self.assertEqual(in_ports[0].count, 1)
1675
1676         def test_flow_multiple_paren(self):
1677             """ test_flow_multiple_paren. """
1678             line = "tunnel(tun_id=0x0,src=192.168.1.1,flags(key)),in_port(2)"
1679             valid = ["tunnel(tun_id=0x0,src=192.168.1.1,flags(key))",
1680                      "in_port(2)"]
1681             rc = flow_line_iter(line)
1682             self.assertEqual(valid, rc)
1683
1684         def test_to_network(self):
1685             """ test_to_network test ipv4_to_network and ipv6_to_network. """
1686             ipv4s = [
1687                 ("192.168.0.1", "192.168.0.1"),
1688                 ("192.168.0.1/255.255.255.255", "192.168.0.1"),
1689                 ("192.168.0.1/255.255.255.0", "192.168.0.0"),
1690                 ("192.168.0.1/255.255.0.0", "192.168.0.0"),
1691                 ("192.168.0.1/255.0.0.0", "192.0.0.0"),
1692                 ("192.168.0.1/0.0.0.0", "0.0.0.0"),
1693                 ("10.24.106.230/255.255.255.255", "10.24.106.230"),
1694                 ("10.24.106.230/255.255.255.0", "10.24.106.0"),
1695                 ("10.24.106.0/255.255.255.0", "10.24.106.0"),
1696                 ("10.24.106.0/255.255.252.0", "10.24.104.0")
1697                 ]
1698
1699             ipv6s = [
1700                 ("1::192:168:0:1", "1::192:168:0:1"),
1701                 ("1::192:168:0:1/1::ffff:ffff:ffff:ffff", "1::192:168:0:1"),
1702                 ("1::192:168:0:1/1::ffff:ffff:ffff:0", "1::192:168:0:0"),
1703                 ("1::192:168:0:1/1::ffff:ffff:0:0", "1::192:168:0:0"),
1704                 ("1::192:168:0:1/1::ffff:0:0:0", "1::192:0:0:0"),
1705                 ("1::192:168:0:1/1::0:0:0:0", "1::"),
1706                 ("1::192:168:0:1/::", "::")
1707                 ]
1708
1709             for (ipv4_test, ipv4_check) in ipv4s:
1710                 self.assertEqual(ipv4_to_network(ipv4_test), ipv4_check)
1711
1712             for (ipv6_test, ipv6_check) in ipv6s:
1713                 self.assertEqual(ipv6_to_network(ipv6_test), ipv6_check)
1714
1715         def test_ui(self):
1716             """ test_ui: test expected ui behavior. """
1717             #pylint: disable=W0212
1718             top_render = Render(80, Render.FIELD_SELECT_TOP)
1719             script_render = Render(80, Render.FIELD_SELECT_SCRIPT)
1720             self.assertEqual(top_render._field_type_select_get(), "in_port")
1721             self.assertEqual(script_render._field_type_select_get(), "all")
1722             #pylint: enable=W0212