[feat] Genesis of cloud2sql

Author: aquamatthias

.gitignore

@@ -27,6 +27,11 @@ share/python-wheels/
 *.egg
 MANIFEST
 
+# IDEs
+.idea/
+.vscode/
+*.iml
+
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

Makefile

@@ -0,0 +1,98 @@
+.PHONY: clean clean-test clean-pyc clean-build clean-env docs help setup
+.DEFAULT_GOAL := help
+.SILENT: clean clean-build clean-pyc clean-test setup
+
+define BROWSER_PYSCRIPT
+import os, webbrowser, sys
+
+from urllib.request import pathname2url
+
+webbrowser.open("file://" + pathname2url(os.path.abspath(sys.argv[1])))
+endef
+export BROWSER_PYSCRIPT
+
+define PRINT_HELP_PYSCRIPT
+import re, sys
+
+for line in sys.stdin:
+	match = re.match(r'^([a-zA-Z_-]+):.*?## (.*)$$', line)
+	if match:
+		target, help = match.groups()
+		print("%-20s %s" % (target, help))
+endef
+export PRINT_HELP_PYSCRIPT
+
+BROWSER := python -c "$$BROWSER_PYSCRIPT"
+
+help:
+	@python -c "$$PRINT_HELP_PYSCRIPT" < $(MAKEFILE_LIST)
+
+clean: clean-build clean-pyc clean-test ## remove all build, test, coverage and Python artifacts
+
+clean-build: ## remove build artifacts
+	rm -fr build/
+	rm -fr out/
+	rm -fr gen/
+	rm -fr dist/
+	rm -fr .eggs/
+	rm -fr .hypothesis/
+	rm -fr .mypy_cache/
+	find . -name '*.egg-info' -exec rm -fr {} +
+	find . -name '*.egg' -exec rm -fr {} +
+
+clean-pyc: ## remove Python file artifacts
+	find . -name '*.pyc' -exec rm -f {} +
+	find . -name '*.pyo' -exec rm -f {} +
+	find . -name '*~' -exec rm -f {} +
+	find . -name '__pycache__' -exec rm -fr {} +
+
+clean-test: ## remove test and coverage artifacts
+	rm -fr .tox/
+	rm -f .coverage
+	rm -fr htmlcov/
+	rm -fr .pytest_cache
+
+clean-env: ## remove environment
+	rm -fr venv
+
+lint: ## static code analysis
+	black --line-length 120 --check cloud2sql tests
+	flake8 cloud2sql
+	mypy --python-version 3.9 --strict cloud2sql tests
+
+test: ## run tests quickly with the default Python
+	pytest
+
+test-all: ## run tests on every Python version with tox
+	tox
+
+coverage: ## check code coverage quickly with the default Python
+	coverage run --source cloud2sql -m pytest
+	coverage combine
+	coverage report -m
+	coverage html
+	$(BROWSER) htmlcov/index.html
+
+venv:
+	python3 -m venv venv --prompt "cloud2sql"
+	. ./venv/bin/activate && python3 -m pip install --upgrade pip
+	# region as long as version 3 is not available on pypi
+	. ./venv/bin/activate && pip install -e ../resoto/resotolib
+	. ./venv/bin/activate && pip install -e ../resoto/plugins/aws
+	. ./venv/bin/activate && pip install -e ../resoto/plugins/digitalocean
+	. ./venv/bin/activate && pip install -e ../resoto/plugins/gcp
+	. ./venv/bin/activate && pip install -e ../resoto/plugins/k8s
+	. ./venv/bin/activate && pip install -e ../resoto/plugins/example_collector
+	# end region
+	. ./venv/bin/activate && pip install -r requirements-test.txt
+	. ./venv/bin/activate && pip install -r requirements.txt
+	. ./venv/bin/activate && pip install -e .
+	. ./venv/bin/activate && mypy --install-types --non-interactive cloud2sql tests
+
+setup: clean clean-env venv
+
+list-outdated:
+	pip list --outdated
+
+install-latest:
+	pip list --outdated --format=freeze | grep -v '^\-e' | cut -d = -f 1  | xargs -n1 pip install -U

cloud2sql/__init__.py

@@ -0,0 +1,13 @@
+"""
+Cloud2Sql
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+Read infrastructure data from your cloud and export it to an SQL database.
+:copyright: © 2022 Some Engineering Inc.
+:license: Apache 2.0, see LICENSE for more details.
+"""
+
+__title__ = "cloud2sql"
+__description__ = "Read infrastructure data from your cloud and export it to an SQL database."
+__author__ = "Some Engineering Inc."
+__license__ = "Apache 2.0"
+__copyright__ = "Copyright © 2022 Some Engineering Inc."

cloud2sql/__main__.py

@@ -0,0 +1,48 @@
+from logging import getLogger
+
+from resotolib.args import Namespace, ArgumentParser
+from resotolib.logger import setup_logger
+from sqlalchemy import create_engine, Engine
+
+from cloud2sql.collect_plugins import collect_from_plugins
+
+log = getLogger("cloud2sql")
+
+
+def parse_args() -> Namespace:
+    parser = ArgumentParser(epilog="Collect data from cloud providers and store it in a database")
+    parser.add_argument("--debug", action="store_true", help="Enable debug logging")
+    parser.add_argument("--config", help="Path to config file", required=True)
+    parser.add_argument(
+        "--show",
+        choices=["progress", "log", "none"],
+        default="progress",
+        help="Output to show during the process. Default: progress",
+    )
+    parser.add_argument(
+        "--db",
+        help="The database url. See https://docs.sqlalchemy.org/en/20/core/engines.html.",
+        required=True,
+    )
+    args = parser.parse_args()
+    args.log_level = "CRITICAL" if args.show != "log" else "DEBUG" if args.debug else "INFO"
+    return args  # type: ignore
+
+
+def collect(engine: Engine, args: Namespace) -> None:
+    try:
+        collect_from_plugins(engine, args)
+    except Exception as e:
+        log.error("Error during collection", e)
+        print(f"Error syncing data to database: {e}")
+
+
+def main() -> None:
+    args = parse_args()
+    setup_logger("cloud2sql", level=args.log_level, force=True)
+    engine = create_engine(args.db)
+    collect(engine, args)
+
+
+if __name__ == "__main__":
+    main()

cloud2sql/collect_plugins.py

@@ -0,0 +1,142 @@
+import concurrent
+import multiprocessing
+from concurrent.futures import ThreadPoolExecutor, Future
+from contextlib import suppress
+from logging import getLogger
+from queue import Queue
+from threading import Event
+from time import sleep
+from typing import Dict, Optional, List, Any
+
+import pkg_resources
+import yaml
+from resotoclient import Kind, Model
+from resotolib.args import Namespace
+from resotolib.baseplugin import BaseCollectorPlugin
+from resotolib.baseresources import BaseResource
+from resotolib.config import Config
+from resotolib.core.actions import CoreFeedback
+from resotolib.core.model_export import node_to_dict
+from resotolib.json import from_json
+from resotolib.proc import emergency_shutdown
+from resotolib.types import Json
+from rich import print as rich_print
+from rich.live import Live
+from sqlalchemy import Engine
+
+from cloud2sql.show_progress import CollectInfo
+from cloud2sql.sql import SqlModel, SqlUpdater
+
+log = getLogger("cloud2sql")
+
+
+def collectors(raw_config: Json, feedback: CoreFeedback) -> Dict[str, BaseCollectorPlugin]:
+    result = {}
+    config: Config = Config  # type: ignore
+    for entry_point in pkg_resources.iter_entry_points("resoto.plugins"):
+        plugin_class = entry_point.load()
+        if issubclass(plugin_class, BaseCollectorPlugin) and plugin_class.cloud in raw_config:
+            log.info(f"Found collector {plugin_class.cloud} ({plugin_class.__name__})")
+            plugin_class.add_config(config)
+            plugin = plugin_class()
+            if hasattr(plugin, "core_feedback"):
+                setattr(plugin, "core_feedback", feedback.with_context(plugin.cloud))
+            result[plugin_class.cloud] = plugin
+
+    Config.init_default_config()
+    Config.running_config.data = {**Config.running_config.data, **Config.read_config(raw_config)}
+    return result
+
+
+def configure(path_to_config: Optional[str]) -> Json:
+    # Config.init_default_config()
+    if path_to_config:
+        with open(path_to_config) as f:
+            return yaml.safe_load(f)  # type: ignore
+    return {}
+
+
+def collect(collector: BaseCollectorPlugin, engine: Engine, feedback: CoreFeedback, args: Namespace) -> None:
+    # collect cloud data
+    feedback.progress_done(collector.cloud, 0, 1)
+    collector.collect()
+    # read the kinds created from this collector
+    kinds = [from_json(m, Kind) for m in collector.graph.export_model(walk_subclasses=False)]
+    model = SqlModel(Model({k.fqn: k for k in kinds}))
+    node_edge_count = len(collector.graph.nodes) + len(collector.graph.edges)
+    ne_current = 0
+    progress_update = 5000
+    feedback.progress_done("sync_db", 0, node_edge_count, context=[collector.cloud])
+    with engine.connect() as conn:
+        # create the ddl metadata from the kinds
+        model.create_schema(conn, args)
+        # ingest the data
+        updater = SqlUpdater(model)
+        node: BaseResource
+        for node in collector.graph.nodes:
+            node._graph = collector.graph
+            exported = node_to_dict(node)
+            exported["type"] = "node"
+            exported["ancestors"] = {
+                "cloud": {"reported": {"id": node.cloud().name}},
+                "account": {"reported": {"id": node.account().name}},
+                "region": {"reported": {"id": node.region().name}},
+                "zone": {"reported": {"id": node.zone().name}},
+            }
+            stmt = updater.insert_node(exported)
+            if stmt is not None:
+                conn.execute(stmt)
+            ne_current += 1
+            if ne_current % progress_update == 0:
+                feedback.progress_done("sync_db", ne_current, node_edge_count, context=[collector.cloud])
+        for from_node, to_node, _ in collector.graph.edges:
+            stmt = updater.insert_node({"from": from_node.chksum, "to": to_node.chksum, "type": "edge"})
+            if stmt is not None:
+                conn.execute(stmt)
+            ne_current += 1
+            if ne_current % progress_update == 0:
+                feedback.progress_done("sync_db", ne_current, node_edge_count, context=[collector.cloud])
+        # commit all the changes to the tmp tables
+        conn.commit()
+    feedback.progress_done(collector.cloud, 1, 1)
+
+
+def show_messages(core_messages: Queue[Json], end: Event) -> None:
+    info = CollectInfo()
+    while not end.is_set():
+        with Live(info.render(), auto_refresh=False, transient=True) as live:
+            with suppress(Exception):
+                info.handle_message(core_messages.get(timeout=1))
+                live.update(info.render())
+    for message in info.rendered_messages():
+        rich_print(message)
+
+
+def collect_from_plugins(engine: Engine, args: Namespace) -> None:
+    # the multiprocessing manager is used to share data between processes
+    mp_manager = multiprocessing.Manager()
+    core_messages: Queue[Json] = mp_manager.Queue()
+    feedback = CoreFeedback("cloud2sql", "collect", "collect", core_messages)
+    raw_config = configure(args.config)
+    all_collectors = collectors(raw_config, feedback)
+    end = Event()
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        try:
+            if args.show == "progress":
+                executor.submit(show_messages, core_messages, end)
+            futures: List[Future[Any]] = []
+            for collector in all_collectors.values():
+                futures.append(executor.submit(collect, collector, engine, feedback, args))
+            for future in concurrent.futures.as_completed(futures):
+                future.result()
+            # when all collectors are done, we can swap all temp tables
+            SqlModel.swap_temp_tables(engine)
+        except Exception as e:
+            # set end and wait for live to finish, otherwise the cursor is not reset
+            end.set()
+            sleep(1)
+            log.error("An error occurred", exc_info=e)
+            print(f"Encountered Error. Giving up. {e}")
+            emergency_shutdown()
+        finally:
+            end.set()

cloud2sql/collect_resoto.py

@@ -0,0 +1,20 @@
+from resotoclient import ResotoClient
+from resotolib.args import Namespace
+from sqlalchemy import Engine
+
+from cloud2sql.sql import SqlModel, SqlUpdater
+
+
+def collect_from_resoto(engine: Engine, args: Namespace) -> None:
+    with ResotoClient("https://localhost:8900", None) as client:
+        model = SqlModel(client.model())
+
+        updater = SqlUpdater(model)
+
+        with engine.connect() as conn:
+            model.create_schema(conn, args)
+            for nd in client.search_graph("id(root) -[0:]->"):
+                stmt = updater.insert_node(nd)
+                if stmt is not None:
+                    conn.execute(stmt)
+            conn.commit()