blob: afbf01b196bd2d5ba418eab635d99d56c06d2667 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
#! /usr/bin/python3
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2020 Intel Corporation
"""
Script to be used with V2 Telemetry.
Allows the user input commands and read the Telemetry response.
"""
import socket
import os
import glob
import json
import readline
# global vars
TELEMETRY_VERSION = "v2"
CMDS = []
def read_socket(sock, buf_len, echo=True):
""" Read data from socket and return it in JSON format """
reply = sock.recv(buf_len).decode()
try:
ret = json.loads(reply)
except json.JSONDecodeError:
print("Error in reply: ", reply)
sock.close()
raise
if echo:
print(json.dumps(ret))
return ret
def handle_socket(path):
""" Connect to socket and handle user input """
sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
global CMDS
print("Connecting to " + path)
try:
sock.connect(path)
except OSError:
print("Error connecting to " + path)
sock.close()
return
json_reply = read_socket(sock, 1024)
output_buf_len = json_reply["max_output_len"]
# get list of commands for readline completion
sock.send("/".encode())
CMDS = read_socket(sock, output_buf_len, False)["/"]
# interactive prompt
text = input('--> ').strip()
while text != "quit":
if text.startswith('/'):
sock.send(text.encode())
read_socket(sock, output_buf_len)
text = input('--> ').strip()
sock.close()
def readline_complete(text, state):
""" Find any matching commands from the list based on user input """
all_cmds = ['quit'] + CMDS
if text:
matches = [c for c in all_cmds if c.startswith(text)]
else:
matches = all_cmds
return matches[state]
readline.parse_and_bind('tab: complete')
readline.set_completer(readline_complete)
readline.set_completer_delims(readline.get_completer_delims().replace('/', ''))
# Path to sockets for processes run as a root user
for f in glob.glob('/var/run/dpdk/*/dpdk_telemetry.%s' % TELEMETRY_VERSION):
handle_socket(f)
# Path to sockets for processes run as a regular user
for f in glob.glob('/run/user/%d/dpdk/*/dpdk_telemetry.%s' %
(os.getuid(), TELEMETRY_VERSION)):
handle_socket(f)
|
