VNFs/DPPD-PROX/helper-scripts/rapid/prox_ctrl.py

   1 ##
   2 ## Copyright (c) 2010-2020 Intel Corporation
   3 ##
   4 ## Licensed under the Apache License, Version 2.0 (the "License");
   5 ## you may not use this file except in compliance with the License.
   6 ## You may obtain a copy of the License at
   7 ##
   8 ##     http://www.apache.org/licenses/LICENSE-2.0
   9 ##
  10 ## Unless required by applicable law or agreed to in writing, software
  11 ## distributed under the License is distributed on an "AS IS" BASIS,
  12 ## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 ## See the License for the specific language governing permissions and
  14 ## limitations under the License.
  15 ##
  16
  17 from __future__ import print_function
  18 from __future__ import division
  19
  20 from builtins import map
  21 from builtins import range
  22 from past.utils import old_div
  23 from builtins import object
  24 import os
  25 import time
  26 import subprocess
  27 import socket
  28 from rapid_log import RapidLog
  29
  30 class prox_ctrl(object):
  31     def __init__(self, ip, key=None, user=None):
  32         self._ip   = ip
  33         self._key  = key
  34         self._user = user
  35         self._proxsock = []
  36
  37     def ip(self):
  38         return self._ip
  39
  40     def test_connect(self):
  41         """Simply try to run 'true' over ssh on remote system.
  42         On failure, raise RuntimeWarning exception when possibly worth
  43         retrying, and raise RuntimeError exception otherwise.
  44         """
  45         return self.run_cmd('test -e /opt/rapid/system_ready_for_rapid', True)
  46
  47     def connect(self):
  48         attempts = 1
  49         RapidLog.debug("Trying to connect to machine \
  50                 on %s, attempt: %d" % (self._ip, attempts))
  51         while True:
  52             try:
  53                 self.test_connect()
  54                 break
  55             except RuntimeWarning as ex:
  56                 RapidLog.debug("RuntimeWarning %d:\n%s"
  57                     % (ex.returncode, ex.output.strip()))
  58                 attempts += 1
  59                 if attempts > 20:
  60                     RapidLog.exception("Failed to connect to instance after %d\
  61                             attempts:\n%s" % (attempts, ex))
  62                     raise Exception("Failed to connect to instance after %d \
  63                             attempts:\n%s" % (attempts, ex))
  64                 time.sleep(2)
  65                 RapidLog.debug("Trying to connect to machine \
  66                        on %s, attempt: %d" % (self._ip, attempts))
  67         RapidLog.debug("Connected to machine on %s" % self._ip)
  68
  69     def connect_socket(self):
  70         attempts = 1
  71         RapidLog.debug("Trying to connect to PROX (just launched) on %s, \
  72                 attempt: %d" % (self._ip, attempts))
  73         sock = None
  74         while True:
  75             sock = self.prox_sock()
  76             if sock is not None:
  77                 break
  78             attempts += 1
  79             if attempts > 20:
  80                 RapidLog.exception("Failed to connect to PROX on %s after %d \
  81                         attempts" % (self._ip, attempts))
  82                 raise Exception("Failed to connect to PROX on %s after %d \
  83                         attempts" % (self._ip, attempts))
  84             time.sleep(2)
  85             RapidLog.debug("Trying to connect to PROX (just launched) on %s, \
  86                     attempt: %d" % (self._ip, attempts))
  87         RapidLog.info("Connected to PROX on %s" % self._ip)
  88         return sock
  89
  90     def close(self):
  91         for sock in self._proxsock:
  92             sock.quit()
  93
  94     def run_cmd(self, command, _connect=False):
  95         """Execute command over ssh on remote system.
  96         Wait for remote command completion.
  97         Return command output (combined stdout and stderr).
  98         _connect argument is reserved for connect() method.
  99         """
 100         cmd = self._build_ssh(command)
 101         try:
 102             return subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 103         except subprocess.CalledProcessError as ex:
 104             #if _connect and ex.returncode == 255:
 105             if _connect:
 106                 raise RuntimeWarning(ex.output.strip())
 107             raise RuntimeError('ssh returned exit status %d:\n%s'
 108                     % (ex.returncode, ex.output.strip()))
 109
 110     def prox_sock(self, port=8474):
 111         """Connect to the PROX instance on remote system.
 112         Return a prox_sock object on success, None on failure.
 113         """
 114         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 115         try:
 116             sock.connect((self._ip, port))
 117             prox = prox_sock(sock)
 118             self._proxsock.append(prox)
 119             return prox
 120         except:
 121             return None
 122
 123     def scp_put(self, src, dst):
 124         """Copy src file from local system to dst on remote system."""
 125         cmd = [ 'scp',
 126                 '-B',
 127                 '-oStrictHostKeyChecking=no',
 128                 '-oUserKnownHostsFile=/dev/null',
 129                 '-oLogLevel=ERROR' ]
 130         if self._key is not None:
 131             cmd.extend(['-i', self._key])
 132         cmd.append(src)
 133         remote = ''
 134         if self._user is not None:
 135             remote += self._user + '@'
 136         remote += self._ip + ':' + dst
 137         cmd.append(remote)
 138         try:
 139             # Actually ignore output on success, but capture stderr on failure
 140             subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 141         except subprocess.CalledProcessError as ex:
 142             raise RuntimeError('scp returned exit status %d:\n%s'
 143                     % (ex.returncode, ex.output.strip()))
 144
 145     def scp_get(self, src, dst):
 146         """Copy src file from remote system to dst on local system."""
 147         cmd = [ 'scp',
 148                 '-B',
 149                 '-oStrictHostKeyChecking=no',
 150                 '-oUserKnownHostsFile=/dev/null',
 151                 '-oLogLevel=ERROR' ]
 152         if self._key is not None:
 153             cmd.extend(['-i', self._key])
 154         remote = ''
 155         if self._user is not None:
 156             remote += self._user + '@'
 157         remote += self._ip + ':/home/' + self._user + src
 158         cmd.append(remote)
 159         cmd.append(dst)
 160         try:
 161             # Actually ignore output on success, but capture stderr on failure
 162             subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 163         except subprocess.CalledProcessError as ex:
 164             raise RuntimeError('scp returned exit status %d:\n%s'
 165                     % (ex.returncode, ex.output.strip()))
 166
 167     def _build_ssh(self, command):
 168         cmd = [ 'ssh',
 169                 '-oBatchMode=yes',
 170                 '-oStrictHostKeyChecking=no',
 171                 '-oUserKnownHostsFile=/dev/null',
 172                 '-oLogLevel=ERROR' ]
 173         if self._key is not None:
 174             cmd.extend(['-i', self._key])
 175         remote = ''
 176         if self._user is not None:
 177             remote += self._user + '@'
 178         remote += self._ip
 179         cmd.append(remote)
 180         cmd.append(command)
 181         return cmd
 182
 183 class prox_sock(object):
 184     def __init__(self, sock):
 185         self._sock = sock
 186         self._rcvd = b''
 187
 188     def __del__(self):
 189         if self._sock is not None:
 190             self._sock.close()
 191             self._sock = None
 192
 193     def start(self, cores):
 194         self._send('start %s' % ','.join(map(str, cores)))
 195
 196     def stop(self, cores):
 197         self._send('stop %s' % ','.join(map(str, cores)))
 198
 199     def speed(self, speed, cores, tasks=[0]):
 200         for core in cores:
 201             for task in tasks:
 202                 self._send('speed %s %s %s' % (core, task, speed))
 203
 204     def reset_stats(self):
 205         self._send('reset stats')
 206
 207     def lat_stats(self, cores, tasks=[0]):
 208         min_lat = 999999999
 209         max_lat = avg_lat = 0
 210         number_tasks_returning_stats = 0
 211         buckets = [0] * 128
 212         self._send('lat all stats %s %s' % (','.join(map(str, cores)),
 213             ','.join(map(str, tasks))))
 214         for core in cores:
 215             for task in tasks:
 216                 stats = self._recv().split(',')
 217             if 'is not measuring' in stats[0]:
 218                 continue
 219             if stats[0].startswith('error'):
 220                 RapidLog.critical("lat stats error: unexpected reply from PROX\
 221                         (potential incompatibility between scripts and PROX)")
 222                 raise Exception("lat stats error")
 223             number_tasks_returning_stats += 1
 224             min_lat = min(int(stats[0]),min_lat)
 225             max_lat = max(int(stats[1]),max_lat)
 226             avg_lat += int(stats[2])
 227             #min_since begin = int(stats[3])
 228             #max_since_begin = int(stats[4])
 229             tsc = int(stats[5]) # Taking the last tsc as the timestamp since
 230                                 # PROX will return the same tsc for each
 231                                 # core/task combination
 232             hz = int(stats[6])
 233             #coreid = int(stats[7])
 234             #taskid = int(stats[8])
 235             mis_ordered = int(stats[9])
 236             extent = int(stats[10])
 237             duplicate = int(stats[11])
 238             stats = self._recv().split(':')
 239             if stats[0].startswith('error'):
 240                 RapidLog.critical("lat stats error: unexpected lat bucket \
 241                         reply (potential incompatibility between scripts \
 242                         and PROX)")
 243                 raise Exception("lat bucket reply error")
 244             buckets[0] = int(stats[1])
 245             for i in range(1, 128):
 246                 stats = self._recv().split(':')
 247                 buckets[i] = int(stats[1])
 248         avg_lat = old_div(avg_lat,number_tasks_returning_stats)
 249         self._send('stats latency(0).used')
 250         used = float(self._recv())
 251         self._send('stats latency(0).total')
 252         total = float(self._recv())
 253         return (min_lat, max_lat, avg_lat, (old_div(used,total)), tsc, hz,
 254                 buckets)
 255
 256     def irq_stats(self, core, bucket, task=0):
 257         self._send('stats task.core(%s).task(%s).irq(%s)' %
 258                 (core, task, bucket))
 259         stats = self._recv().split(',')
 260         return int(stats[0])
 261
 262     def show_irq_buckets(self, core, task=0):
 263         rx = tx = drop = tsc = hz = 0
 264         self._send('show irq buckets %s %s' % (core,task))
 265         buckets = self._recv().split(';')
 266         buckets = buckets[:-1]
 267         return buckets
 268
 269     def core_stats(self, cores, tasks=[0]):
 270         rx = tx = drop = tsc = hz = rx_non_dp = tx_non_dp = tx_fail = 0
 271         self._send('dp core stats %s %s' % (','.join(map(str, cores)),
 272             ','.join(map(str, tasks))))
 273         for core in cores:
 274             for task in tasks:
 275                 stats = self._recv().split(',')
 276                 if stats[0].startswith('error'):
 277                     if stats[0].startswith('error: invalid syntax'):
 278                         RapidLog.critical("dp core stats error: unexpected \
 279                                 invalid syntax (potential incompatibility \
 280                                 between scripts and PROX)")
 281                         raise Exception("dp core stats error")
 282                     continue
 283                 rx += int(stats[0])
 284                 tx += int(stats[1])
 285                 rx_non_dp += int(stats[2])
 286                 tx_non_dp += int(stats[3])
 287                 drop += int(stats[4])
 288                 tx_fail += int(stats[5])
 289                 tsc = int(stats[6])
 290                 hz = int(stats[7])
 291         return rx, rx_non_dp, tx, tx_non_dp, drop, tx_fail, tsc, hz
 292
 293     def multi_port_stats(self, ports=[0]):
 294         rx = tx = port_id = tsc = no_mbufs = errors = 0
 295         self._send('multi port stats %s' % (','.join(map(str, ports))))
 296         result = self._recv().split(';')
 297         if result[0].startswith('error'):
 298             RapidLog.critical("multi port stats error: unexpected invalid \
 299                     syntax (potential incompatibility between scripts and \
 300                     PROX)")
 301             raise Exception("multi port stats error")
 302         for statistics in result:
 303             stats = statistics.split(',')
 304             port_id = int(stats[0])
 305             rx += int(stats[1])
 306             tx += int(stats[2])
 307             no_mbufs += int(stats[3])
 308             errors += int(stats[4])
 309             tsc = int(stats[5])
 310         return rx, tx, no_mbufs, errors, tsc
 311
 312     def set_random(self, cores, task, offset, mask, length):
 313         self._send('set random %s %s %s %s %s' % (','.join(map(str, cores)),
 314             task, offset, mask, length))
 315
 316     def set_size(self, cores, task, pkt_size):
 317         self._send('pkt_size %s %s %s' % (','.join(map(str, cores)), task,
 318             pkt_size))
 319
 320     def set_imix(self, cores, task, imix):
 321         self._send('imix %s %s %s' % (','.join(map(str, cores)), task,
 322             ','.join(map(str,imix))))
 323
 324     def set_value(self, cores, task, offset, value, length):
 325         self._send('set value %s %s %s %s %s' % (','.join(map(str, cores)),
 326             task, offset, value, length))
 327
 328     def quit_prox(self):
 329         self._send('quit')
 330
 331     def _send(self, cmd):
 332         """Append LF and send command to the PROX instance."""
 333         if self._sock is None:
 334             raise RuntimeError("PROX socket closed, cannot send '%s'" % cmd)
 335         self._sock.sendall(cmd.encode() + b'\n')
 336
 337     def _recv(self):
 338         """Receive response from PROX instance, return it with LF removed."""
 339         if self._sock is None:
 340             raise RuntimeError("PROX socket closed, cannot receive anymore")
 341         pos = self._rcvd.find(b'\n')
 342         while pos == -1:
 343             self._rcvd += self._sock.recv(256)
 344             pos = self._rcvd.find(b'\n')
 345         rsp = self._rcvd[:pos]
 346         self._rcvd = self._rcvd[pos+1:]
 347         return rsp.decode()