summaryrefslogtreecommitdiffstats
path: root/anta/cli/exec/commands.py
blob: 9842cc88276bbbcf929fa94eb49d8ea9c4ff97e7 (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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
# Copyright (c) 2023-2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""Click commands to execute various scripts on EOS devices."""

from __future__ import annotations

import asyncio
import logging
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING

import click
from yaml import safe_load

from anta.cli.console import console
from anta.cli.exec.utils import clear_counters_utils, collect_commands, collect_scheduled_show_tech
from anta.cli.utils import inventory_options

if TYPE_CHECKING:
    from anta.inventory import AntaInventory

logger = logging.getLogger(__name__)


@click.command
@inventory_options
def clear_counters(inventory: AntaInventory, tags: set[str] | None) -> None:
    """Clear counter statistics on EOS devices."""
    asyncio.run(clear_counters_utils(inventory, tags=tags))


@click.command()
@inventory_options
@click.option(
    "--commands-list",
    "-c",
    help="File with list of commands to collect",
    required=True,
    show_envvar=True,
    type=click.Path(file_okay=True, dir_okay=False, exists=True, readable=True, path_type=Path),
)
@click.option(
    "--output",
    "-o",
    show_envvar=True,
    type=click.Path(file_okay=False, dir_okay=True, exists=False, writable=True, path_type=Path),
    help="Directory to save commands output.",
    default=f"anta_snapshot_{datetime.now(tz=timezone.utc).astimezone().strftime('%Y-%m-%d_%H_%M_%S')}",
    show_default=True,
)
def snapshot(inventory: AntaInventory, tags: set[str] | None, commands_list: Path, output: Path) -> None:
    """Collect commands output from devices in inventory."""
    console.print(f"Collecting data for {commands_list}")
    console.print(f"Output directory is {output}")
    try:
        with commands_list.open(encoding="UTF-8") as file:
            file_content = file.read()
            eos_commands = safe_load(file_content)
    except FileNotFoundError:
        logger.error("Error reading %s", commands_list)
        sys.exit(1)
    asyncio.run(collect_commands(inventory, eos_commands, output, tags=tags))


@click.command()
@inventory_options
@click.option(
    "--output",
    "-o",
    default="./tech-support",
    show_default=True,
    help="Path for test catalog",
    type=click.Path(path_type=Path),
    required=False,
)
@click.option(
    "--latest",
    help="Number of scheduled show-tech to retrieve",
    type=int,
    required=False,
)
@click.option(
    "--configure",
    help="Ensure devices have 'aaa authorization exec default local' configured (required for SCP on EOS). THIS WILL CHANGE THE CONFIGURATION OF YOUR NETWORK.",
    default=False,
    is_flag=True,
    show_default=True,
)
def collect_tech_support(
    inventory: AntaInventory,
    tags: set[str] | None,
    output: Path,
    latest: int | None,
    *,
    configure: bool,
) -> None:
    """Collect scheduled tech-support from EOS devices."""
    asyncio.run(collect_scheduled_show_tech(inventory, output, configure=configure, tags=tags, latest=latest))