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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146 | import argparse
import logging
from rich import box
from rich.console import Console
from rich.table import Table
from virl2_client import ClientLibrary
URL = "https://10.0.0.1"
USER = "admin"
PASS = "pass"
def parse_args():
parser = argparse.ArgumentParser(
description="Display CML links or bring selected links up/down."
)
parser.add_argument(
"-w",
"--workspace",
help="Display links only in this CML workspace (Workspace ID).",
)
parser.add_argument(
"--link-id",
nargs="+",
metavar="LINK_ID",
help="One or more Link IDs to operate on.",
)
action = parser.add_mutually_exclusive_group()
action.add_argument(
"--down", action="store_true", help="Bring selected links down."
)
action.add_argument("--up", action="store_true", help="Bring selected links up.")
args = parser.parse_args()
if (args.down or args.up) and not args.link_id:
parser.error("--link-id is required with --down or --up")
if args.link_id and not (args.down or args.up):
parser.error("--link-id can only be used with --down or --up")
return args
def print_links(labs, workspace=None):
"""Print links, optionally limited to one workspace."""
title = "CML All Links" if workspace is None else f"CML Links: {workspace}"
table = Table(title=title, box=box.MINIMAL_DOUBLE_HEAD)
for column in (
"Lab",
"Link ID",
"Node-A",
"Node-B",
"State",
"Bandwidth",
"Latency",
"Loss",
"Jitter",
):
table.add_column(column)
links = [
(lab, link)
for lab in labs
if workspace is None or lab.id == workspace
for link in lab.links()
]
links.sort(
key=lambda item: (
f"{item[1].node_a.label} {item[1].interface_a.label}".casefold(),
f"{item[1].node_b.label} {item[1].interface_b.label}".casefold(),
item[0].title.casefold(),
item[1].id.casefold(),
)
)
synced_labs = set()
for lab, link in links:
if lab.id not in synced_labs:
lab.sync_states()
synced_labs.add(lab.id)
is_up = str(link.state).upper() == "STARTED"
link_state = "Up" if is_up else "-"
condition = link.get_condition()
values = [
str(condition.get("bandwidth", "")),
str(condition.get("latency", "")),
str(condition.get("loss", "")),
str(condition.get("jitter", "")),
]
table.add_row(
lab.title,
link.id,
f"{link.node_a.label} {link.interface_a.label}",
f"{link.node_b.label} {link.interface_b.label}",
link_state,
*values,
style="white" if is_up else "grey50",
)
Console().print(table)
def change_links(labs, link_ids, action, workspace=None):
"""Start or stop configured links in the selected workspace."""
links_by_id = {
link.id: link
for lab in labs
if workspace is None or lab.id == workspace
for link in lab.links()
}
missing = [link_id for link_id in link_ids if link_id not in links_by_id]
if missing:
scope = f" in workspace {workspace!r}" if workspace else ""
raise SystemExit("Link IDs not found" + scope + ": " + ", ".join(missing))
for link_id in link_ids:
if action == "down":
links_by_id[link_id].stop()
else:
links_by_id[link_id].start()
def main():
args = parse_args()
logging.getLogger("virl2_client.virl2_client").addFilter(
_SuppressSSLVerificationWarning()
)
cml = ClientLibrary(URL, USER, PASS, ssl_verify=False)
cml.is_system_ready(wait=True)
labs = cml.all_labs()
action = "down" if args.down else "up" if args.up else None
if action is not None:
change_links(labs, args.link_id, action, args.workspace)
labs = cml.all_labs()
print_links(labs, args.workspace)
class _SuppressSSLVerificationWarning(logging.Filter):
"""Suppress only the warning emitted for the intentionally disabled SSL check."""
def filter(self, record):
return record.getMessage() != "SSL Verification disabled"
if __name__ == "__main__":
main()
|