sonic-trimming.py

"""
This CLI plugin was auto-generated by using 'sonic-cli-gen' utility
"""

import click
import utilities_common.cli as clicommon

from sonic_py_common import logger
from utilities_common.switch_trimming import (
    CFG_SWITCH_TRIMMING,
    STATE_SWITCH_CAPABILITY,
    STATE_CAP_TRIMMING_CAPABLE_KEY,
    STATE_CAP_QUEUE_MODE_KEY,
    STATE_CAP_QUEUE_MODE_DYNAMIC,
    STATE_CAP_QUEUE_MODE_STATIC,
    CFG_TRIM_QUEUE_INDEX_DYNAMIC,
    CFG_TRIM_KEY,
    STATE_CAP_KEY,
    UINT32_MAX,
    UINT8_MAX,
    SYSLOG_IDENTIFIER,
    get_db,
    to_str,
)


log = logger.Logger(SYSLOG_IDENTIFIER)
log.set_min_log_priority_info()


#
# Validators ----------------------------------------------------------------------------------------------------------
#


class SizeTypeValidator(click.ParamType):
    """ Size option validator """
    name = "integer"

    def convert(self, value, param, ctx):
        click.IntRange(0, UINT32_MAX).convert(value, param, ctx)
        return value


class DscpTypeValidator(click.ParamType):
    """ Dscp option validator """
    name = "integer"

    def convert(self, value, param, ctx):
        click.IntRange(0, UINT8_MAX).convert(value, param, ctx)
        return value


class QueueTypeValidator(click.ParamType):
    """ Queue index option validator """
    name = "text"

    def get_metavar(self, param):
        db = get_db(click.get_current_context())

        entry = db.get_all(db.STATE_DB, "{}|{}".format(STATE_SWITCH_CAPABILITY, STATE_CAP_KEY))
        entry.setdefault(STATE_CAP_QUEUE_MODE_KEY, "N/A")

        cap_list = entry[STATE_CAP_QUEUE_MODE_KEY].split(',')

        if cap_list.count(STATE_CAP_QUEUE_MODE_DYNAMIC) == len(cap_list):
            return "dynamic"
        elif cap_list.count(STATE_CAP_QUEUE_MODE_STATIC) == len(cap_list):
            return "INTEGER"

        return "[INTEGER|dynamic]"

    def convert(self, value, param, ctx):
        db = get_db(ctx)

        entry = db.get_all(db.STATE_DB, "{}|{}".format(STATE_SWITCH_CAPABILITY, STATE_CAP_KEY))

        entry.setdefault(STATE_CAP_TRIMMING_CAPABLE_KEY, "false")
        entry.setdefault(STATE_CAP_QUEUE_MODE_KEY, "N/A")

        if entry[STATE_CAP_TRIMMING_CAPABLE_KEY] == "false":
            raise click.UsageError("Failed to configure {}: operation is not supported".format(
                param.get_error_hint(ctx)), ctx
            )

        if not entry[STATE_CAP_QUEUE_MODE_KEY]:
            raise click.UsageError("Failed to configure {}: no queue resolution mode capabilities".format(
                param.get_error_hint(ctx)), ctx
            )

        verify_cap = True

        if entry[STATE_CAP_QUEUE_MODE_KEY] == "N/A":
            verify_cap = False

        cap_list = entry[STATE_CAP_QUEUE_MODE_KEY].split(',')

        if value == CFG_TRIM_QUEUE_INDEX_DYNAMIC:
            if verify_cap and (STATE_CAP_QUEUE_MODE_DYNAMIC not in cap_list):
                self.fail("dynamic queue resolution mode is not supported", param, ctx)
        else:
            if verify_cap and (STATE_CAP_QUEUE_MODE_STATIC not in cap_list):
                self.fail("static queue resolution mode is not supported", param, ctx)

            click.IntRange(0, UINT8_MAX).convert(value, param, ctx)

        return value


#
# DB interface --------------------------------------------------------------------------------------------------------
#


def update_entry_validated(db, table, key, data, create_if_not_exists=False):
    """ Update entry in table and validate configuration.
    If attribute value in data is None, the attribute is deleted.

    Args:
        db (swsscommon.ConfigDBConnector): Config DB connector object.
        table (str): Table name to add new entry to.
        key (Union[str, Tuple]): Key name in the table.
        data (Dict): Entry data.
        create_if_not_exists (bool):
            In case entry does not exists already a new entry
            is not created if this flag is set to False and
            creates a new entry if flag is set to True.
    Raises:
        Exception: when cfg does not satisfy YANG schema.
    """

    cfg = db.get_config()
    cfg.setdefault(table, {})

    if not data:
        raise click.ClickException(f"No field/values to update {key}")

    if create_if_not_exists:
        cfg[table].setdefault(key, {})

    if key not in cfg[table]:
        raise click.ClickException(f"{key} does not exist")

    entry_changed = False
    for attr, value in data.items():
        if value == cfg[table][key].get(attr):
            continue
        if value is not None:
            cfg[table][key][attr] = value
            entry_changed = True

    if not entry_changed:
        return

    db.set_entry(table, key, cfg[table][key])


#
# CLI -----------------------------------------------------------------------------------------------------------------
#


@click.group(
    name="switch-trimming",
    cls=clicommon.AliasedGroup
)
def SWITCH_TRIMMING():
    """ Configure switch trimming feature """

    pass


@SWITCH_TRIMMING.command(
    name="global"
)
@click.option(
    "-s", "--size", "size",
    help="Configures size (in bytes) to trim eligible packet",
    type=SizeTypeValidator(),
)
@click.option(
    "-d", "--dscp", "dscp",
    help="Configures DSCP value assigned to a packet after trimming",
    type=DscpTypeValidator(),
)
@click.option(
    "-q", "--queue", "queue",
    help="Configures queue index to use for transmission of a packet after trimming",
    type=QueueTypeValidator(),
)
@clicommon.pass_db
@click.pass_context
def SWITCH_TRIMMING_GLOBAL(ctx, db, size, dscp, queue):
    """ Configure switch trimming global """

    if not (size or dscp or queue):
        raise click.UsageError("Failed to configure switch trimming global: no options are provided", ctx)

    table = CFG_SWITCH_TRIMMING
    key = CFG_TRIM_KEY

    data = {
        "size": size,
        "dscp_value": dscp,
        "queue_index": queue,
    }

    try:
        update_entry_validated(db.cfgdb, table, key, data, create_if_not_exists=True)
        log.log_notice("Configured switch trimming global: {}".format(to_str(data)))
    except Exception as e:
        log.log_error("Failed to configure switch trimming global: {}".format(str(e)))
        ctx.fail(str(e))


def register(cli):
    """ Register new CLI nodes in root CLI.

    Args:
        cli: Root CLI node.
    Raises:
        Exception: when root CLI already has a command
                   we are trying to register.
    """
    cli_node = SWITCH_TRIMMING
    if cli_node.name in cli.commands:
        raise Exception(f"{cli_node.name} already exists in CLI")
    cli.add_command(SWITCH_TRIMMING)