#!/usr/bin/env python
# ex: set shiftwidth=4 tabstop=4 expandtab:
# Copyright C 2002-2008 by Isilon Systems, Inc. All rights not expressly
# granted herein are reserved.
# Created in the United States of America.
#
# This digital media is protected by U.S. and international copyright and
# intellectual property laws. Unless otherwise specified, all information
# and screens appearing as part of this digital medium, including
# software, services, documents, text, images, icons, and logos design;
# the selection, assembly, arrangement, and design thereof; and the code
# that enables its presentation, are the sole property of Isilon Systems,
# Inc.
#
# Except as otherwise required by applicable law, no part of any
# copyrighted material may be reproduced, modified, translated,
# distributed, published, or transmitted by any means, electronic or
# mechanical, for any purpose, without the prior written consent of Isilon
# Systems.
#
# Isilon and Isilon Systems are registered trademarks of Isilon Systems,
# Inc.  "Isilon IQ" is a trademark of Isilon Systems, Inc.  All other
# trademarks and registered trademarks are the property of their
# respective holders.

import getopt
import os
import random
import re
import sys
import time

try:
    import isi.ph.perf as perf
    import isi.ph.throughputtest as tput
    import isi.ph.util as util
    from isi.ph.util import err, ParseStandardOpts
except ImportError:
    try:
        # old-style, run directly out of the ph directory
        import perf as perf
        import throughputtest as tput
        import util as util
        from util import err, ParseStandardOpts
    except ImportError:
        print "Either run on a node or from within the ph library directory."
        sys.exit(1)

def Version():
    return "$Revision: 207696 $".split(" ")[1]

def usage():
    print """
usage: %(prog)s [opts] node_ip1 [...] clients client_ip1 [...]

          %(prog)s scales read/write load against an array using given clients
          via various client access protocols.

          -b:  Blocksize to use; can use units like KB, TB, GiB; default 512KB
          -c:  Get performance collector output
          -d:  Minimum time to sample for
          -D:  Sampling numerator.  Sample happens for
                max(sampling_min, -D/nthread)
       --dir:  Directory to use under mount point; defaults to ph_dir
    --direct:  Open files with O_DIRECT. Linux 2.6 NFS tests only.
          -e:  Random read exponent.
          -h:  This message
          -i:  Thread increment
--ignore-versions: Ignore PH version mismatches among test machines
    --layout:  Drive layout; more often, use --pattern.  Takes # of disks.
--mount-options:  Mount options to use on the clients, e.g.
                "wsize=32768,rsize=32768,intr,nfsvers=3,tcp"
                If the nfsvers or tcp/udp are not chosen, will try to work it
                out from the protocol.  For BSD clients, specify all options.
          -n:  Number of runs to perform for each configuration
      --nics:  Number of external nics to use per node.
          -p:  Protection level for test (e.g. 3x or +9)
     --pcdir:  Directory to store pcdata in; default /ifs/pc_dir.
    --pctime:  Length of PC collection period, in seconds; default 20.
      --port:  Server port
          -P:  Protocols to use (nfsv3tcp, cifs, nfsv3udp, nfsv2, http, ftp)
   --pattern:  Access pattern ('streaming' or 'default'/'concurrency')
          -q:  Only run test against dnas present.  Error if none present.
          -s:  Starting thread count (defaults to 1)
          -S:  Filesize to use; can use units like KB, TB, GiB, e.g. 3.5GiB
      --sync:  Open files for writing with O_SYNC or FILE_FLAG_WRITE_THROUGH
      --seed:  Seed RNG for random read test.
          -t:  Maximal thread count
       --tag:  Tag the test run with a short string
          -T:  Tests to run.  Valid tests are:
                cached_read, uncached_read, random_read, coalesced_write,
                uncoalesced_write, coalesced_overwrite, uncoalesced_overwrite
          -u:  Username; defaults to os.environ['LOGNAME']
          -v:  Verbose mode
   --verbose:  Verbose mode
   --version:  Show versions of all loaded ph modules
          -w:  Minimum time to warmup
          -W:  Numerator for warmup.  Warmup = max(warmup_min, -W/nthread)
          -z:  Number of storage nodes to remove from group during test.

Streaming read options (size options can use units, eg KB, GiB):
    --buffer:  Buffersize for streaming reader test.
 --framerate:  Frames per second in streaming test.  Set -S to frame size.
 --framesize:  Framesize, only needed if streaming from a large single file.
    --single:  Stream from a single large file.  Use -S to set filesize.
               This option implies many frames in a single file, like a movie.

Sample command:

%(prog)s -Tcoalesced_overwrite -n 1 -i 1 -t 4 -Pcifs -d 3 -w 10 -S 3GiB \\
    10.53.145.1 10.53.145.2 10.53.145.3 clients 10.8.5.87

# Sample streaming read test: 1 frame per file, many files:
%(prog)s --framesize 12MB --framerate 30 -t40 -T streaming_read -Pnfsv3tcp \\
    10.53.145.1 10.53.145.2 10.53.145.3 clients tbox588

# Sample streaming read test: 1 file per thread, many frames per 4GB file:
%(prog)s --single -S 4GB --framesize 12MB --framerate 30 -t40 -Pnfsv3tcp \\
    -T streaming_read 10.53.145.1 10.53.145.2 10.53.145.3 clients tbox588

""" % ({ 'prog' : sys.argv[0].split("/")[-1] })
    sys.exit(1)

readtests = ["cached_read",
             "uncached_read",
             "cached_degraded_read",
             "uncached_degraded_read",
             "random_read",
             "streaming_read"]

writetests = ["coalesced_write",
              "uncoalesced_write",
              "coalesced_overwrite",
              "uncoalesced_overwrite",
              "streaming_write"]
alltests = readtests + writetests

# User configurable variables
threadspernode = 3
degraded = 0
threadincrement = 1

# time controls
sampling_numerator = 10
sampling_min = 1
warmup_numerator = 3
warmup_min = .5
pc_dur = 20

access_pattern = None
blocksize = 512 * 1024      # TODO need to differ defaults by protocol
degrade_level = 0
dnas_only = 0
drive_layout = None
filesize = None
ignore_versions = False
username = None
mountopts = None
nic_count = None
numruns = 1
collect_pc = False
pcbase = "/ifs/pc_dir"
pc_id = None
protattr = (util.ATTR_FEC, 0, 1)
protection = "+1"
protos = ["nfsv3tcp"]
randomexp = 0
startnthreads = 1
write_sync = False
directio = False
tag = None
tests = alltests
totalthreads = None
tpstats_base = None
verbose = False
workingdir = 'ph_dir'

# streaming read options
buffersize = 500 * 1024 * 1024
framerate = 0
framesize = 0
single_file = False

if len(sys.argv) == 1:
    usage()

flags, others = getopt.getopt(sys.argv[1:],
                        "b:cd:D:e:hi:n:p:P:qs:S:t:T:u:vW:w:z:",
                        [ "buffer=", "dir=", "direct", "framerate=",
                          "framesize=", "help", "ignore-versions", "layout=",
                          "mount-options=", "nics=", "pattern=",
                          "pcdir=", "pctime=", "port=", "seed=", "single",
                          "sync", "tag=", "tpstats", "verbose", "version", ])
for opt in flags:
    flag = opt[0]
    if flag == "-b":
        blocksize = util.ConvertSizeUnits(opt[1])
        if blocksize is None:
            err("Unable to convert blocksize '%s'" % opt[1])
    elif flag == "--buffer":
        buffersize = util.ConvertSizeUnits(opt[1])
        if buffersize is None:
            err("Unable to convert buffersize '%s'" % opt[1])
    elif flag == "-c":
        collect_pc = True
    elif flag == "-d":
        sampling_min = float(opt[1])
    elif flag == "-D":
        sampling_numerator = float(opt[1])
    elif flag == "--dir":
        workingdir = opt[1]
    elif flag == "--direct":
        directio = True
    elif flag == "-e":
        randomexp = float(opt[1])
        print "nonsequential access e =", randomexp
    elif flag == "--framerate":
        framerate = int(opt[1])
    elif flag == "--framesize":
        framesize = util.ConvertSizeUnits(opt[1])
        if framesize is None:
            err("Unable to convert framesize '%s'" % opt[1])
    elif flag in ("-h", "--help"):
        usage()
    elif flag == "-i":
        threadincrement = int(opt[1])
    elif flag == "--ignore-versions":
        ignore_versions = True
    elif flag == "--layout":
        drive_layout = int(opt[1])
    elif flag == "--mount-options":
        mountopts = opt[1]
    elif flag == "-n":
        numruns= int(opt[1])
    elif flag == "--nics":
        nic_count = int(opt[1])
    elif flag == "-o":
        official = 1
    elif flag == "-p":
        protattr = util.MakeProtection(opt[1])
        protection = opt[1]
    elif flag == "-P":
        protos = opt[1].split(",")
        for proto in protos:
            if proto not in ("local", "nfsv3tcp", "nfsv3udp", "nfsv2", "cifs",
                             "http", "ftp", "hdfs"):
                err(0, "Unrecognized protocol", proto)
    elif flag == "--pattern":
        if opt[1] == 'streaming':
            access_pattern = util.ATTR_PATTERN_STREAM
        elif opt[1] in ('default', 'concurrency'):
            access_pattern = util.ATTR_PATTERN_DEF
        else:
            err("Only streaming and default patterns supported")
    elif flag == "--pcdir":
        pcbase = opt[1]
    elif flag == "--pctime":
        pc_dur = float(opt[1])
    elif flag == "--port":
        util.SetServerPort(opt[1])
    elif flag == "-q":
        dnas_only = 1
    elif flag == "-s":
        startnthreads = int(opt[1])
    elif flag == "-S":
        filesize = util.ConvertSizeUnits(opt[1])
        if filesize is None:
            err("Unable to convert filesize '%s'" % opt[1])
    elif flag == "--seed":
        seed = int(opt[1])
        print "random seed found, value", seed
        random.seed(seed)
    elif flag == "--sync":
        write_sync = True
    elif flag == "--single":
        single_file = True
    elif flag == "-t":
        totalthreads = int(opt[1])
    elif flag == "--tag":
        tag = opt[1]
    elif flag == "-T":
        tests = opt[1].split(",")
        for test in tests:
            if test not in alltests:
                err(0, test, "not in", alltests)
    elif flag == "--tpstats":
        tpstats_base = "/ifs/.ifsvar/modules/ph/tests/%d" % int(time.time())
    elif flag == "-u":
        username = opt[1]
        if re.search(r"[()$!; \\/]", username):
            err("Special characters not allowed in username: '%s'" % username)
    elif flag in ("-v", "--verbose"):
        verbose = True
    elif flag == "--version":
        util.PrintVersions("isi_ph_ovt", Version())
    elif flag == "-w":
        warmup_min = float(opt[1])
    elif flag == "-W":
        warmup_numerator = float(opt[1])
    elif flag == "-z":
        degrade_level = int(opt[1])


# check streaming read options
if 'streaming_read' in tests or 'streaming_write' in tests:
    # require filesize for streaming single file test because using a default
    # would be confusing
    if single_file:
        if not filesize or not framesize:
            err("single file mode needs file size, frame size and frame rate")
        if filesize < 3*1024**3:
            err("a filesize less than 3GB will end up cached")
    else:
        # filesize is same as framesize in multi-file test, so be adaptable:
        if framesize and not filesize:
            filesize = framesize
        elif filesize and not framesize:
            framesize = filesize
        else:
            # but warn about confusion
            err("in multi-file stream test, give only framesize or filesize")

    # individual frames > 1GB are surely an error
    if framesize > 1024**3:
        err("unbelievable framesize: %d" % framesize)

    if not framerate:
        err("frame rate required for streaming tests")

    if framesize > buffersize:
        err("buffer smaller than frame size")

# set default filesize
if filesize is None:
    filesize = 3 * 1024L * 1024 * 1024

# this limitation results from the version of apache that comes with onefs
if 'http' in protos:
    if filesize >= 2 * 1024**3:
        err("http tests limited to files of less than 2GB")

# try to correct mountopts based on selected protocol
mountopts = util.FixMountOpts(protos, mountopts)

# parse out the clients and nodes, and look up the nodes to see which are dnas
if not others:
    print "Need nodes for all tests."
    usage()

(clients, selected_node_ips,
            selected_dna_ips, sysctls) = ParseStandardOpts(others)
if sysctls != [()]:
    err("Sysctl ramps not supported in OVT")
sysctls = ()

if dnas_only and not selected_dna_ips:
    print "-q flag set but no dnas selected!"
    usage()

if (len(protos) > 1 or protos[0] != 'local') and not clients:
    print "Need clients for non-local protocol tests."
    usage()

# selected_nodes
# selected_dnas
# all_selected_nodes
#   1 ip per node selected by the user or by GetNodeIpsByInterfaceCount.
#   _nodes only has storage; _dnas only dnas; all_ has both
#
# selected_node_ips
# selected_dna_ips
# all_selected_ips
#   Any number of IPs per node.
#
# selected_node_dict
#   Keys: all_selected_nodes; values: lists of IPs from all_selected_ips.
#
# cluster:
#   Keys: LNNs; values: lists of IPs.  Includes *all* nodes and external IPs.

# This may not return all the nodes in the cluster, if selected_node_ips
# or selected_dna_ips was only a selection.
try:
    selected_nodes, selected_dnas, selected_node_dict, cluster = \
            util.GetUniqueNodes(selected_node_ips, selected_dna_ips)

    if nic_count:
        selected_node_ips, selected_dna_ips, selected_node_dict = \
            util.GetNodeIpsByInterfaceCount(selected_nodes, selected_dnas,
                                            nic_count, cluster)
except Exception, e:
    print "Error organizing node IPs: %s" % str(e)
    sys.exit(1)

all_selected_ips   = selected_node_ips + selected_dna_ips
all_selected_nodes = selected_nodes + selected_dnas

if dnas_only:
    print "Will use these IPs for mounts:", ", ".join(selected_dna_ips)
else:
    print "Will use these IPs for mounts:", ", ".join(all_selected_ips)

try:
    unameinfo = util.GetUnameInfo(all_selected_nodes, clients)
    nodeconf  = util.GetNodeConf(all_selected_nodes)
except Exception, e:
    print "Error getting node information: %s" % str(e)
    sys.exit(1)

if totalthreads == None:
    if dnas_only:
        totalthreads = len(selected_dnas) * threadspernode
    else:
        totalthreads = len(all_selected_nodes) * threadspernode

# Check that all machines in the test have the same ph module versions
# Always test the versions; AllVersionsMatch prints some useful information.
if not util.AllVersionsMatch(all_selected_nodes, clients) and \
        not ignore_versions:
    print "PH versions do not match; use --ignore-versions to ignore"
    sys.exit(1)

# the user's name (or the -l option) set the name of the test runner in the DB
if not username:
    if 'LOGNAME' in os.environ:
        username = os.environ["LOGNAME"]
    else:
        print "LOGNAME not in environment and -l not given"
        usage()

# any attempt to set drive layout above 0 sets PATTERN_STREAM automatically
if drive_layout and access_pattern == util.ATTR_PATTERN_DEF:
    err("No drive layout accepted with default/concurrency pattern.")
layout_attrs = (access_pattern, drive_layout)

# pick an ID for perf collection so all perf results for a set
# of tests can go in a single subdirectory of data/pcdata
if collect_pc and pc_id is None:
    pc_id = "%s.%s.%s" % (os.uname()[1], username, repr(time.time()))

# TODO should this even be done in OVT?  Need to reset values certainly;
#      should also consider setting dirty ratio, background ratio, etc
#error = util.SetupClientProcEntries(clients, util.stdprocentries)
#if error:
#    err(error, "SetupClientProcEntries")

timecontrols = (warmup_numerator, warmup_min,
                    sampling_numerator, sampling_min, pc_dur)

setupread = 0

for test in tests:
    for proto in protos:
        total_tp = 0
        ss_tp = 0
        tpstats = None

        if tpstats_base:
            tpstats = perf.TpPickler(tpstats_base, proto, test, numruns,
                                     all_selected_nodes, clients,
                                     startnthreads, totalthreads,
                                     threadincrement)

        for run in xrange(numruns):
            if tpstats:
                tpstats.startRun(run)

            # Create the working directory on the nodes
            # create an entire dataset tree per node
            if not setupread and test in readtests:
                attrs = protattr + (util.ATTR_COAL,) + layout_attrs
                if test == "streaming_read":
                    error = tput.PopulateStreamFiles(workingdir,
                                                     all_selected_nodes,
                                                     startnthreads,
                                                     totalthreads,
                                                     filesize, extattr=attrs,
                                                     frame_rate=framerate,
                                                     single_file=single_file)
                else:
                    error = tput.PopulateFiles(workingdir, all_selected_nodes,
                                               startnthreads, totalthreads,
                                               filesize, attrs)
                setupread = 1

                if error:
                    err(error, "setting up read test %s" % test)

            # test_nodes: one ip per node, of nodes actively used in test
            # test_ips: all ips to be tested for the nodes in test_nodes
            if degrade_level:
                degrade_cmds, test_nodes, test_ips = tput.CreateDegradeCommand(
                                                        selected_nodes,
                                                        selected_node_dict,
                                                        degrade_level)
                test_nodes += selected_dnas
                test_ips   += selected_dna_ips
            else:
                test_nodes = all_selected_nodes
                test_ips   = all_selected_ips

            if dnas_only:
                test_nodes = selected_dnas
                test_ips   = selected_dna_ips

            if tpstats:
                tpstats.addRunNodes(test_nodes, test_ips)

            if test in readtests:
                if test == "cached_read":
                    error, tpdata = tput.CachedScaledReadTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, blocksize, 2000 * 1024L * 1024L,
                        filesize=filesize, mountopts=mountopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)
                elif test == "uncached_read":
                    error, tpdata = tput.UncachedScaledReadTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, blocksize,
                        filesize=filesize, mountopts=mountopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)
                elif test == "random_read":
                    error, tpdata = tput.RandomReadTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, blocksize, randomexp=randomexp,
                        filesize=filesize, mountopts=mountopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)
                elif test == "streaming_read":
                    error, tpdata = tput.StreamingReadTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, filesize, blocksize, buffersize,
                        framerate, framesize, single_file=single_file,
                        mountopts=mountopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)
                else:
                    err("Unknown test type", test)
            else:
                # start write tests
                if test == "uncoalesced_write" or \
                       test == "uncoalesced_overwrite":
                    coalesced = util.ATTR_NONCOAL
                else:
                    coalesced = util.ATTR_COAL

                attrs = protattr + (coalesced,) + layout_attrs

                fileopts = {}
                if write_sync:
                    fileopts['sync'] = True
                elif directio:
                    fileopts['direct'] = True

                # prepare files for write test
                if test == "coalesced_overwrite" or \
                       test == "uncoalesced_overwrite":
                    overwrite = 1

                    error = tput.PopulateFiles(workingdir + "/writes",
                                               all_selected_nodes,
                                               startnthreads, totalthreads,
                                               filesize, attrs)
                else:
                    overwrite = 0
                    if test == "streaming_write":
                        error = tput.PopulateStreamFiles(workingdir,
                                    all_selected_nodes, startnthreads,
                                    totalthreads, 0, extattr=attrs,
                                    frame_rate=framerate,
                                    single_file=single_file)
                    else:
                        error = tput.PopulateFiles(
                                    workingdir + "/writes",
                                    all_selected_nodes, startnthreads,
                                    totalthreads, 0, attrs)

                if error:
                    err(error, "setting up write test %s" % test)

                # run write test
                if test == "streaming_write":
                    error, tpdata = tput.StreamingWriteTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, filesize, blocksize, buffersize,
                        framerate, framesize, single_file=single_file,
                        mountopts=mountopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)
                else:
                    error, tpdata = tput.WriteTest(
                        workingdir, test_ips, test_nodes,
                        all_selected_nodes, clients, run, sysctls,
                        startnthreads, totalthreads, threadincrement,
                        timecontrols, blocksize,
                        coalesced=coalesced, overwrite=overwrite,
                        filesize=filesize, mountopts=mountopts,
                        fileopts=fileopts,
                        verbose=verbose, pc_id=pc_id, pcbase=pcbase,
                        tpstats=tpstats, proto=proto)

            if degrade_level:
                tput.EndDegradeCommands(degrade_cmds)

            if error:
                err(error, test)

            if tpstats:
                tpstats.stopRun()

            node_count = len(test_nodes)
            thread_counts = tpdata.keys()
            thread_counts.sort()
            for tcnt in thread_counts:
                nodetp = 0.0
                for node in all_selected_nodes:
                    node_key = "node_%s" % node
                    if node_key in tpdata[tcnt]:
                        nodetp += tpdata[tcnt][node_key][0]

                clienttp = 0.0
                for client in clients:
                    client_key = "client_%s" % client
                    if client_key in tpdata[tcnt]:
                        for cmdtp, extra in tpdata[tcnt][client_key]:
                            clienttp += cmdtp

                nodetp_mbit   = float(nodetp) / 1000 / 1000 * 8
                clienttp_mbit = float(clienttp) / 1000 / 1000 * 8

                if clients:
                    cdisplay = "  clients %7.1f Mbit/s " % clienttp_mbit
                else:
                    cdisplay = ""

                s = tcnt != 1 and "s:" or ": "
                print "%3d thread%s %s   nodes %7.1f Mbit/s" % \
                        (tcnt, s, cdisplay, nodetp_mbit)

            print

        if tpstats:
            tpstats.stopProtocolTest()
            print "Throughput statistics: " + tpstats.tpfile

