7e39e23f64
And example script for getting and setting OVS external-ids. Since currently there is no nmcli support for these properties yet, the script becomes more interesting. This "example" is rather long, and it showcases less the usage of libnm (which is rather trivial, with respect to configuring NMSettingOvsExternalIDs). Instead, it aims to provide a useful command line tool for debugging. Hence, it's mostly concerned with an elaborate command line syntax and useful print output.
426 lines
12 KiB
Python
Executable File
426 lines
12 KiB
Python
Executable File
#!/usr/bin/env python
|
|
# SPDX-License-Identifier: GPL-2.0+
|
|
#
|
|
# Copyright (C) 2017, 2020 Red Hat, Inc.
|
|
#
|
|
|
|
#
|
|
# set and show OVS external-ids for a connection:
|
|
#
|
|
|
|
import sys
|
|
import os
|
|
import re
|
|
import pprint
|
|
|
|
import gi
|
|
|
|
gi.require_version("NM", "1.0")
|
|
from gi.repository import GLib, NM
|
|
|
|
MODE_GET = "get"
|
|
MODE_SET = "set"
|
|
|
|
|
|
def pr(v):
|
|
pprint.pprint(v, indent=4, depth=5, width=60)
|
|
|
|
|
|
HAS_LIBNM_DEBUG = os.getenv("LIBNM_CLIENT_DEBUG") is not None
|
|
|
|
|
|
def _print(msg=""):
|
|
if HAS_LIBNM_DEBUG:
|
|
# we want to use the same logging mechanism as libnm's debug
|
|
# logging with "LIBNM_CLIENT_DEBUG=trace,stdout".
|
|
NM.utils_print(0, msg + "\n")
|
|
return
|
|
print(msg)
|
|
|
|
|
|
def mainloop_run(timeout_msec=0, mainloop=None):
|
|
if mainloop is None:
|
|
mainloop = GLib.MainLoop()
|
|
|
|
timeout_id = None
|
|
timeout_reached = []
|
|
|
|
if timeout_msec > 0:
|
|
|
|
def _timeout_cb(unused):
|
|
# it can happen that the caller already quit the mainloop
|
|
# otherwise. In that case, we don't want to signal a timeout.
|
|
if mainloop.is_running():
|
|
timeout_reached.append(1)
|
|
mainloop.quit()
|
|
return True
|
|
|
|
timeout_id = GLib.timeout_add(timeout_msec, _timeout_cb, None)
|
|
|
|
mainloop.run()
|
|
if timeout_id:
|
|
GLib.source_remove(timeout_id)
|
|
return not timeout_reached
|
|
|
|
|
|
def usage():
|
|
_print("%s g[et] PROFILE [ GETTER ]" % (sys.argv[0]))
|
|
_print("%s s[et] [--test] PROFILE SETTER" % (sys.argv[0]))
|
|
_print(
|
|
" PROFILE := [id | uuid | type] STRING | [ ~id | ~type ] REGEX_STRING | STRING"
|
|
)
|
|
_print(" GETTER := ( KEY | ~REGEX_KEY ) [... GETTER]")
|
|
_print(" SETTER := ( + | - | -KEY | [+]KEY VALUE ) [... SETTER]")
|
|
|
|
|
|
def die(msg, show_usage=False):
|
|
_print("FAILED: %s" % (msg))
|
|
if show_usage:
|
|
usage()
|
|
sys.exit(1)
|
|
|
|
|
|
def die_usage(msg):
|
|
die(msg, show_usage=True)
|
|
|
|
|
|
def parse_args(argv):
|
|
had_dash_dash = False
|
|
args = {
|
|
"mode": MODE_GET,
|
|
"profile_arg": None,
|
|
"ids_arg": [],
|
|
"do_test": False,
|
|
}
|
|
i = 1
|
|
while i < len(argv):
|
|
a = argv[i]
|
|
|
|
if i == 1:
|
|
if a in ["s", "set"]:
|
|
args["mode"] = MODE_SET
|
|
elif a in ["g", "get"]:
|
|
args["mode"] = MODE_GET
|
|
else:
|
|
die_usage("unexpected mode argument '%s'" % (a))
|
|
i += 1
|
|
continue
|
|
|
|
if a == "--test":
|
|
args["do_test"] = True
|
|
i += 1
|
|
continue
|
|
|
|
if args["profile_arg"] is None:
|
|
if a in ["id", "~id", "uuid", "type", "~type"]:
|
|
if i + 1 >= len(argv):
|
|
die_usage("'%s' requires an argument'" % (a))
|
|
args["profile_arg"] = (a, argv[i + 1])
|
|
i += 2
|
|
continue
|
|
|
|
if a == "*":
|
|
a = None
|
|
args["profile_arg"] = ("*", a)
|
|
i += 1
|
|
continue
|
|
|
|
if args["mode"] == MODE_GET:
|
|
args["ids_arg"].append(a)
|
|
i += 1
|
|
continue
|
|
|
|
if not a:
|
|
die_usage("argument should specify a external-id but is empty string")
|
|
|
|
if a[0] == "-":
|
|
v = (a, None)
|
|
i += 1
|
|
elif a == "+":
|
|
v = (a, None)
|
|
i += 1
|
|
else:
|
|
if a[0] != "+":
|
|
a = "+" + a
|
|
if i + 1 >= len(argv):
|
|
die_usage("'%s' requires an argument'" % (a))
|
|
v = (a, argv[i + 1])
|
|
i += 2
|
|
|
|
args["ids_arg"].append(v)
|
|
|
|
if args["mode"] == MODE_SET:
|
|
if not args["ids_arg"]:
|
|
die_usage("Requires one or more external-ids to set or delete")
|
|
|
|
return args
|
|
|
|
|
|
def connection_to_str(connection, show_type=False):
|
|
if show_type:
|
|
return "%s (%s, %s)" % (
|
|
connection.get_id(),
|
|
connection.get_uuid(),
|
|
connection.get_connection_type(),
|
|
)
|
|
return "%s (%s)" % (connection.get_id(), connection.get_uuid())
|
|
|
|
|
|
def connections_filter(connections, profile_arg):
|
|
connections = list(sorted(connections, key=connection_to_str))
|
|
if not profile_arg:
|
|
return connections
|
|
# we preserve the order of the selected connections. And
|
|
# if connections are selected multiple times, we return
|
|
# them multiple times.
|
|
l = []
|
|
f = profile_arg
|
|
for c in connections:
|
|
if f[0] == "id":
|
|
if f[1] == c.get_id():
|
|
l.append(c)
|
|
elif f[0] == "~id":
|
|
if re.match(f[1], c.get_id()):
|
|
l.append(c)
|
|
elif f[0] == "uuid":
|
|
if f[1] == c.get_uuid():
|
|
l.append(c)
|
|
elif f[0] == "type":
|
|
if f[1] == c.get_connection_type():
|
|
l.append(c)
|
|
elif f[0] == "~type":
|
|
if re.match(f[1], c.get_connection_type()):
|
|
l.append(c)
|
|
else:
|
|
assert f[0] == "*"
|
|
if f[1] is None:
|
|
l.append(c)
|
|
else:
|
|
if f[1] in [c.get_uuid(), c.get_id()]:
|
|
l.append(c)
|
|
return l
|
|
|
|
|
|
def ids_select(ids, mode, ids_arg):
|
|
ids = list(ids)
|
|
if not ids_arg:
|
|
return (ids, [])
|
|
|
|
keys = set()
|
|
requested = []
|
|
for d in ids_arg:
|
|
if mode == MODE_GET:
|
|
if d[0] == "~":
|
|
r = re.compile(d[1:])
|
|
keys.update([k for k in ids if r.match(k)])
|
|
else:
|
|
keys.update([k for k in ids if k == d])
|
|
if d not in requested:
|
|
requested.append(d)
|
|
else:
|
|
d2 = d[0]
|
|
assert d2[0] in ["-", "+"]
|
|
d3 = d2[1:]
|
|
if d3 in ids:
|
|
keys.add(d3)
|
|
return (list([k for k in ids if k in keys]), requested)
|
|
|
|
|
|
def connection_print(connection, mode, ids_arg, dbus_path, prefix=""):
|
|
sett = connection.get_setting(NM.SettingOvsExternalIDs)
|
|
|
|
if sett is not None:
|
|
all_ids = list(sett.get_data_keys())
|
|
keys, requested = ids_select(all_ids, mode, ids_arg)
|
|
num_str = "%s" % (len(all_ids))
|
|
else:
|
|
keys = []
|
|
requested = []
|
|
num_str = "none"
|
|
|
|
_print(
|
|
"%s%s [%s]" % (prefix, connection_to_str(connection, show_type=True), num_str)
|
|
)
|
|
_print("%s %s" % (prefix, dbus_path))
|
|
if sett is not None:
|
|
dd = sett.get_property(NM.SETTING_OVS_EXTERNAL_IDS_DATA)
|
|
else:
|
|
dd = {}
|
|
for k in keys:
|
|
v = sett.get_data(k)
|
|
assert v is not None
|
|
assert v == dd.get(k, None)
|
|
_print('%s "%s" = "%s"' % (prefix, k, v))
|
|
for k in requested:
|
|
_print('%s "%s" = <unset>' % (prefix, k))
|
|
|
|
|
|
def do_get(connections, ids_arg):
|
|
first_line = True
|
|
for c in connections:
|
|
if first_line:
|
|
first_line = False
|
|
else:
|
|
_print()
|
|
connection_print(c, MODE_GET, ids_arg, dbus_path=c.get_path())
|
|
|
|
|
|
def do_set(nmc, connection, ids_arg, do_test):
|
|
|
|
remote_connection = connection
|
|
connection = NM.SimpleConnection.new_clone(remote_connection)
|
|
|
|
connection_print(
|
|
connection, MODE_SET, [], remote_connection.get_path(), prefix="BEFORE: "
|
|
)
|
|
_print()
|
|
|
|
sett = connection.get_setting(NM.SettingOvsExternalIDs)
|
|
|
|
for d in ids_arg:
|
|
op = d[0][0]
|
|
key = d[0][1:]
|
|
val = d[1]
|
|
|
|
oldval = None
|
|
if sett is not None:
|
|
oldval = sett.get_data(key)
|
|
|
|
if op == "-":
|
|
assert val is None
|
|
if key == "":
|
|
if sett is None:
|
|
_print(" DEL: setting (ovs-external-ids group was not present)")
|
|
else:
|
|
connection.remove_setting(NM.SettingOvsExternalIDs)
|
|
sett = None
|
|
_print(" DEL: setting")
|
|
continue
|
|
|
|
if sett is None:
|
|
_print(' DEL: "%s" (ovs-external-ids group was not present)' % (key))
|
|
continue
|
|
if oldval is None:
|
|
_print(' DEL: "%s" (id was unset)' % (key))
|
|
continue
|
|
_print(' DEL: "%s" (id was set to"%s")' % (key, oldval))
|
|
sett.set_data(key, None)
|
|
continue
|
|
|
|
if key == "":
|
|
assert val is None
|
|
if sett is None:
|
|
sett = NM.SettingOvsExternalIDs.new()
|
|
connection.add_setting(sett)
|
|
_print(" SET: setting (external-ids group was added)")
|
|
continue
|
|
|
|
_print(" SET: setting (external-ids group was present)")
|
|
continue
|
|
|
|
assert val is not None
|
|
|
|
if sett is None:
|
|
sett = NM.SettingOvsExternalIDs.new()
|
|
connection.add_setting(sett)
|
|
_print(
|
|
' SET: "%s" = "%s" (external-ids group was not present)' % (key, val)
|
|
)
|
|
elif oldval is None:
|
|
_print(' SET: "%s" = "%s" (new)' % (key, val))
|
|
elif oldval != val:
|
|
_print(' SET: "%s" = "%s" (was "%s")' % (key, val, oldval))
|
|
else:
|
|
_print(' SET: "%s" = "%s" (unchanged)' % (key, val))
|
|
sett.set_data(key, val)
|
|
|
|
if do_test:
|
|
_print()
|
|
_print("Only show. Run without --test to set")
|
|
return
|
|
|
|
mainloop = GLib.MainLoop()
|
|
result_error = []
|
|
|
|
def callback(c, result):
|
|
try:
|
|
c.update2_finish(result)
|
|
except Exception as e:
|
|
result_error.append(e)
|
|
mainloop.quit()
|
|
|
|
remote_connection.update2(
|
|
connection.to_dbus(NM.ConnectionSerializationFlags.ALL),
|
|
NM.SettingsUpdate2Flags.NO_REAPPLY,
|
|
None,
|
|
None,
|
|
callback,
|
|
)
|
|
|
|
mainloop_run(mainloop=mainloop)
|
|
|
|
if result_error:
|
|
_print()
|
|
_print("FAILURE to commit connection: %s" % (result_error[0]))
|
|
return
|
|
|
|
# NMClient received the completion of Update2() call. It also received
|
|
# a property changed signal that the profile changed, and it is about
|
|
# to fetch the new value. However, that value is not yet here.
|
|
#
|
|
# libnm should provide a better API for this. For example, not signal
|
|
# completion of update2() until the profile was refetched. Or, indicate
|
|
# that the settings are dirty, so we would know how long to wait.
|
|
#
|
|
# Add an ugly workaround here and wait a bit.
|
|
_print()
|
|
_print("WORKAROUND: wait for connection to change")
|
|
mainloop_run(timeout_msec=500)
|
|
|
|
if remote_connection is not nmc.get_object_by_path(remote_connection.get_path()):
|
|
_print()
|
|
_print(
|
|
"Connection %s no longer exists after commit"
|
|
% (remote_connection.get_path())
|
|
)
|
|
return
|
|
|
|
_print()
|
|
connection_print(
|
|
remote_connection, MODE_SET, [], remote_connection.get_path(), prefix="AFTER: "
|
|
)
|
|
|
|
_print()
|
|
if remote_connection.compare(connection, NM.SettingCompareFlags.EXACT):
|
|
_print("resulting connection is as expected")
|
|
else:
|
|
_print("WARNING: resulting connection is not as expected")
|
|
|
|
|
|
###############################################################################
|
|
|
|
if __name__ == "__main__":
|
|
|
|
args = parse_args(sys.argv)
|
|
|
|
nmc = NM.Client.new(None)
|
|
|
|
connections = connections_filter(nmc.get_connections(), args["profile_arg"])
|
|
|
|
if args["mode"] == MODE_SET:
|
|
if len(connections) != 1:
|
|
_print(
|
|
"To set the external-ids of a connection, exactly one connection must be selected via id|uuid. Instead, %s connection matched ([%s])"
|
|
% (
|
|
len(connections),
|
|
", ".join([connection_to_str(c) for c in connections]),
|
|
)
|
|
)
|
|
die_usage("Select unique connection to set")
|
|
do_set(nmc, connections[0], args["ids_arg"], do_test=args["do_test"])
|
|
else:
|
|
if len(connections) < 1:
|
|
_print("No connection selected for printing the external ids")
|
|
die_usage("Select connection to get")
|
|
do_get(connections, args["ids_arg"])
|