tools: create a new tooling repo

services/tools/cli/.gitignore

@@ -0,0 +1,72 @@
+# Byte-compiled files
+__pycache__/
+*.py[cod]
+
+# Distribution / packaging
+.Python
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+*.egg-info/
+dist/
+build/
+
+# IDE and editor files
+.idea/
+.vscode/
+*.swp
+*.swo
+*.sublime-project
+*.sublime-workspace
+
+# Jupyter Notebook checkpoints
+.ipynb_checkpoints
+
+# Python virtual environments
+.venv/
+venv/
+ENV/
+env/
+
+# Test and coverage reports
+.coverage
+*.cover
+*.log
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+*.egg
+*.egg-info/
+dist/
+build/
+*.whl
+
+# Pytest cache
+.pytest_cache/
+
+# MyPy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Pylint
+pylint-report.txt
+pylint-output.txt
+
+# Other files
+*.DS_Store
+*.db
+*.sqlite3
+*.sqlite
+*.log
+*.pot
+*.mo
+*.pyc
+*.pyo
+*.pyd

services/tools/cli/args_utils.py

@@ -0,0 +1,18 @@
+from argparse import ArgumentTypeError
+import os
+
+def get_full_path(relative_path):
+    if not os.path.isabs(relative_path):
+        current_dir = os.getcwd()
+        full_path = os.path.join(current_dir, relative_path)
+    else:
+        full_path = relative_path
+    
+    return full_path
+
+def dir_arg_type(path):
+    full_path = get_full_path(path) 
+    if os.path.isdir(full_path):
+        return full_path
+    else:
+        raise ArgumentTypeError(f"{path} nor {full_path} are not a valid path")

services/tools/cli/base.py

@@ -0,0 +1,21 @@
+import sys
+import logging
+
+class BaseCLi:
+    def __init__(self):
+        self.logger = logging.getLogger(self.__class__.__name__)
+        handler = logging.StreamHandler()
+        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+        handler.setFormatter(formatter)
+        self.logger.addHandler(handler)
+        self.logger.setLevel(logging.ERROR)
+
+    def info(self, message):
+        self.logger.info(message)
+
+    def warning(self, message):
+        self.logger.warning(message)
+
+    def error(self, message, exit_code=1):
+        self.logger.error(message)
+        sys.exit(exit_code)

services/tools/cli/constants.py

@@ -0,0 +1,46 @@
+PORTAINER_HOST_URLS = {
+    'vm-tools-100-55': 'https://portainer55.davydovcloud.com',
+    'vm-mixed-100-98': 'https://portainer98.davydovcloud.com'
+}
+
+PORTAINER_STACK_CLI_ACTIONS = {
+    'LIST': {
+        'value': 'list',
+        'arg_required': False,
+        'arg_name': '',
+    },
+    'CREATE': {
+        'value': 'create',
+        'arg_required': True,
+        'arg_name': 'stack-dir-path',
+    },
+    'REMOVE': {
+        'value': 'remove',
+        'arg_required': True,
+        'arg_name': 'stack-id',
+    },
+    'REDEPLOY': {
+        'value': 'redeploy',
+        'arg_required': True,
+        'arg_name': 'stack-id',
+    },
+    'START': {
+        'value': 'start',
+        'arg_required': True,
+        'arg_name': 'stack-id',
+    },
+    'STOP': {
+        'value': 'stop',
+        'arg_required': True,
+        'arg_name': 'stack-id',
+    },
+    'INFO': {
+        'value': 'info',
+        'arg_required': True,
+        'arg_name': 'stack-id',
+    },
+}
+
+DEFAULT_PORTAINER_JWT_CONFIG_FILE = '.portainer_cli_config'
+FUNCTIONAL_USER_USERNAME_ENV = 'PORTAINER_USERNAME'
+FUNCTIONAL_USER_PASSWORD_ENV = 'PORTAINER_PASSWORD'

services/tools/cli/portainer_host.py

@@ -0,0 +1,76 @@
+import json
+from base import BaseCLi
+from constants import (
+    DEFAULT_PORTAINER_JWT_CONFIG_FILE,
+    FUNCTIONAL_USER_USERNAME_ENV,
+    FUNCTIONAL_USER_PASSWORD_ENV,
+)
+import os
+import requests
+
+class MissingEnvironmentVariable(Exception):
+    pass
+
+class EmptyConfigException(Exception):
+    pass
+
+class PortainerHost(BaseCLi):
+
+    def __init__(self, host_url):
+        super().__init__()
+        self.host_url = host_url
+        self.api_url = f'{host_url}/api'
+        self.jwt_config = []
+        home_directory = os.path.expanduser("~")
+        config_file_path = os.path.join(home_directory, DEFAULT_PORTAINER_JWT_CONFIG_FILE)
+        try:
+            self.read_current_config(config_file_path)
+        except EmptyConfigException:
+            self.create_config(config_file_path)
+
+    def read_current_config(self, file_path):
+        try:
+            with open(file_path, 'r') as file:
+                self.jwt_config = json.load(file)
+
+        except FileNotFoundError:
+            print(f"Configuration does not exist. Creating...")
+            raise EmptyConfigException()
+        
+        hostnames = [config['hostname'] for config in self.jwt_config]
+        if self.host_url not in hostnames:
+            raise EmptyConfigException()
+            
+    def create_config(self, file_path):
+        config = {
+            "hostname": self.host_url,
+            "jwt": self.get_jwt_token(),
+        }
+        self.jwt_config.append(config)
+        with open(file_path, 'w') as file:
+            json.dump(self.jwt_config, file, indent=4)
+
+    def get_jwt_token(self):
+        def get_env(env_key):
+            try:
+                return os.environ[env_key]
+            except KeyError:
+                raise MissingEnvironmentVariable(f"{env_key} env does not exist")
+
+        payload = {
+            "Username": get_env(FUNCTIONAL_USER_USERNAME_ENV),
+            "Password": get_env(FUNCTIONAL_USER_PASSWORD_ENV)
+        }
+        auth_endpoint = f'{self.api_url}/auth'
+        try:
+            response = requests.post(
+                auth_endpoint,
+                json=payload
+            )
+            if response.status_code != 200:
+                self.error(f'Return code {response.status_code} from {auth_endpoint} with message {response.text}')
+        except requests.exceptions.RequestException as e:
+            self.error("An error occurred:", e)
+        data = response.json()
+        return data['jwt']
+

services/tools/cli/portainer_stack.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+
+from argparse import ArgumentParser
+from args_utils import dir_arg_type
+from base import BaseCLi
+from constants import PORTAINER_HOST_URLS, PORTAINER_STACK_CLI_ACTIONS
+from portainer_host import PortainerHost
+
+CLI_DESCRIPTION='''
+This cli is meant to be used to execute whoel LCM flow of
+Portainer host stacks. Portainer host can be supplied separately,
+or by the tag.
+'''
+
+class PortainerStackCLi(BaseCLi):
+
+    def configure_args(self, parser: ArgumentParser):
+        parser.add_argument(
+            '-a',
+            '--action',
+            help='Action to be performed on Portainer host',
+            required=True,
+            choices=[action['value'] for action in PORTAINER_STACK_CLI_ACTIONS.values()]
+        )
+        parser.add_argument(
+            '-s',
+            '--stack-dir-path',
+            help='Directory with the stack related files',
+            type=dir_arg_type
+        )
+        parser.add_argument(
+            '-i',
+            '--stack-id',
+            help='Id of the stack in Portainer (see list)',
+            type=int
+        )
+        parser.add_argument(
+            '-n',
+            '--hostname',
+            help='Portainer hostname to be used',
+            required=True,
+            type=str,
+            choices=PORTAINER_HOST_URLS.keys()
+        )
+
+
+    def process_args(self, parser: ArgumentParser):
+        args = parser.parse_args()
+        required_args = {
+            'stack-dir-path': args.stack_dir_path,
+            'stack-id': args.stack_id,
+        }
+
+        for action in PORTAINER_STACK_CLI_ACTIONS.values():
+            if not (args.action == action['value'] and action['arg_required']):
+                continue
+
+            for arg_name, arg_value in required_args.items():
+                if action['arg_name'] == arg_name and not arg_value:
+                    parser.print_help()
+                    self.error(f'Argument --{arg_name} is required.')
+        return args
+
+    def __init__(self):
+        super().__init__()
+        parser = ArgumentParser(description=CLI_DESCRIPTION)
+        self.configure_args(parser)
+        self.args = self.process_args(parser)
+
+    def start(self):
+        host = PortainerHost(PORTAINER_HOST_URLS[self.args.hostname])
+        
+
+def main():
+    cli = PortainerStackCLi()
+    cli.start()
+
+if __name__ == "__main__":
+    main()