History: Store the history of an Azure tenancy

.flake8

@@ -1,3 +1,5 @@
 [flake8]
 ignore = W191,W503,E203,E741
 max-line-length = 240
+per-file-ignores =
+	**/conftest.py:F811

.github/workflows/pants.yml

@@ -46,7 +46,7 @@ jobs:
         env:
           integration_test_secrets: ${{secrets.integration_test_secrets}}
       - name: Upload test coverage to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
           files: dist/coverage/python/coverage.xml

llamazure/azgraph/integration_test.py

@@ -1,37 +1,19 @@
 """Integration test against a real, live Azure"""
 # pylint: disable=redefined-outer-name
 
-import os
-from typing import Any
-
 import pytest
-import yaml
-from azure.identity import ClientSecretCredential
 
 from llamazure.azgraph.azgraph import Graph
 from llamazure.azgraph.models import Req, Res, ResErr
-
-
-def print_output(name: str, output: Any):
-	should_print = os.environ.get("INTEGRATION_PRINT_OUTPUT", "False") == "True"
-	if should_print:
-		print(name, output)
+from llamazure.test.credentials import load_credentials
+from llamazure.test.inspect import print_output
 
 
 @pytest.fixture()
 @pytest.mark.integration
 def graph():
 	"""Run integration test"""
-
-	secrets = os.environ.get("integration_test_secrets")
-	if not secrets:
-		with open("cicd/secrets.yml", mode="r", encoding="utf-8") as f:
-			secrets = f.read()
-	secrets = yaml.safe_load(secrets)
-	client = secrets["azgraph"]
-
-	credential = ClientSecretCredential(tenant_id=client["tenant"], client_id=client["appId"], client_secret=client["password"])
-
+	credential = load_credentials()
 	g = Graph.from_credential(credential)
 	return g
 

llamazure/history/BUILD

@@ -0,0 +1,34 @@
+python_sources(
+	name="history",
+)
+
+python_tests(
+	name="tests",
+)
+
+python_distribution(
+	name="llamazure.history",
+	dependencies=[":history"],
+	long_description_path="llamazure/history/readme.md",
+	provides=python_artifact(
+		name="llamazure.history",
+		version="0.0.1",
+		description="Build a history of an Azure tenancy",
+		author="Daniel Goldman",
+		classifiers=[
+			"Development Status :: 3 - Alpha",
+			"Programming Language :: Python :: 3.8",
+			"Programming Language :: Python :: 3.9",
+			"Programming Language :: Python :: 3.10",
+			"Programming Language :: Python :: 3.11",
+			"Topic :: Utilities",
+			"Topic :: Internet :: Log Analysis",
+		],
+		license="Round Robin 2.0.0",
+		long_description_content_type="text/markdown",
+	),
+)
+
+python_test_utils(
+	name="test_utils",
+)

llamazure/history/app.py

@@ -0,0 +1,112 @@
+"""The llamazure.history application, a webserver to collect and present the history of Azure tenancies"""
+from __future__ import annotations
+
+import datetime
+from typing import Generator, List, Optional, TypeVar
+from uuid import UUID
+
+from azure.identity import DefaultAzureCredential
+from fastapi import Depends, FastAPI
+from pydantic import BaseModel
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from llamazure.azgraph import azgraph
+from llamazure.history.collect import Collector, CredentialCache
+from llamazure.history.data import DB, TSDB, Res
+
+T = TypeVar("T")
+Provider = Generator[T, None, None]
+
+
+class CredentialCacheDefault(CredentialCache):
+	"""Load Azure credentials with default loader"""
+
+	@staticmethod
+	def credential():
+		"""Get the Default Azure credential"""
+		return DefaultAzureCredential()
+
+	def azgraph(self, tenant_id: UUID) -> azgraph.Graph:
+		return azgraph.Graph.from_credential(self.credential())
+
+
+class Settings(BaseSettings):
+	"""Settings for llamazure.history"""
+
+	model_config = SettingsConfigDict(env_nested_delimiter="__")
+
+	class DB(BaseModel):
+		"""Settings for DB"""
+
+		connstr: str
+
+	db: Settings.DB
+
+
+settings = Settings()  # type: ignore
+
+
+def get_collector() -> Provider[Collector]:
+	"""FastAPI Dependency for Collector"""
+	yield Collector(
+		CredentialCacheDefault(),
+		DB(TSDB(settings.db.connstr)),
+	)
+
+
+def get_db() -> Provider[DB]:
+	"""FastAPI Dependency for DB"""
+	yield DB(TSDB(settings.db.connstr))
+
+
+app = FastAPI()
+
+
+@app.post("/collect/snapshots")
+async def collect_snapshot(tenant_id: UUID, collector: Collector = Depends(get_collector)):
+	"""Dispatch the collection of a snapshot"""
+	collector.take_snapshot(tenant_id)
+
+
+@app.post("/collect/delta")
+async def collect_delta(tenant_id: UUID, delta: dict, collector: Collector = Depends(get_collector)):
+	"""Insert a single delta"""
+	collector.insert_deltas(tenant_id, [delta])
+
+
+@app.post("/collect/deltas")
+async def collect_deltas(tenant_id: UUID, deltas: List[dict], collector: Collector = Depends(get_collector)):
+	"""Insert multiple deltas"""
+	collector.insert_deltas(tenant_id, deltas)
+
+
+@app.get("/history")
+async def read_history(at: Optional[datetime.datetime] = None, db: DB = Depends(get_db)) -> Res:
+	"""Read history at a point in time"""
+	if at is None:
+		return db.read_latest()
+	else:
+		return db.read_at(at)
+
+
+@app.get("/resources/{resource_id}")
+async def read_resource(
+	resource_id: str,
+	ti: Optional[datetime.datetime] = None,
+	tf: Optional[datetime.datetime] = None,
+	db: DB = Depends(get_db),
+) -> Res:
+	return db.read_resource(resource_id, ti, tf)
+
+
+@app.get("/ping")
+async def ping() -> str:
+	"""PING this service for up check"""
+	now = datetime.datetime.now(datetime.timezone.utc).isoformat()
+	return f"PONG {now}"
+
+
+@app.post("/admin/init_db")
+async def init_db(db: DB = Depends(get_db)):
+	"""Initialize the tables in the database"""
+	db.create_tables()

llamazure/history/collect.py

@@ -0,0 +1,63 @@
+import datetime
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Dict, Generator, List, Tuple, cast
+from uuid import UUID
+
+from llamazure.azgraph import azgraph
+from llamazure.history.data import DB
+from llamazure.rid import mp
+from llamazure.tresource.mp import MPData, TresourceMPData
+
+
+def reformat_resources_for_tresource(resources):
+	"""Reformat mp_resources for TresourceMPData"""
+	for r in resources:
+		path, azobj = mp.parse(r["id"])
+		mpdata = MPData(azobj, r)
+		yield path, mpdata
+
+
+def reformat_resources_for_db(tree: TresourceMPData) -> Generator[Tuple[str, Dict], None, None]:
+	return ((cast(str, path), mpdata.data) for path, mpdata in tree.resources.items() if mpdata.data is not None)
+
+
+class CredentialCache(ABC):
+	@abstractmethod
+	def azgraph(self, tenant_id: UUID) -> azgraph.Graph:
+		"""Get the azgraph.Graph instance for this tenant"""
+
+
+@dataclass
+class Collector:
+	"""Load data from Azure Resource Manager and put into the DB"""
+
+	credentials: CredentialCache
+	db: DB
+
+	def take_snapshot(self, tenant_id: UUID):
+		"""Take a snapshot and insert it into the DB"""
+		resources = self.credentials.azgraph(tenant_id).q("Resources")
+		if isinstance(resources, azgraph.ResErr):
+			raise RuntimeError(azgraph.ResErr)
+
+		tree: TresourceMPData[Dict] = TresourceMPData()
+		tree.add_many(reformat_resources_for_tresource(resources))
+
+		self.db.insert_snapshot(
+			time=self.snapshot_time(),
+			azure_tenant=tenant_id,
+			resources=reformat_resources_for_db(tree),
+		)
+
+	def insert_deltas(self, tenant_id: UUID, deltas: List[Dict]):
+		tree: TresourceMPData[Dict] = TresourceMPData()
+		tree.add_many(reformat_resources_for_tresource(deltas))
+
+		request_snapshot_time = self.snapshot_time()
+		for rid, data in reformat_resources_for_db(tree):
+			self.db.insert_delta(time=request_snapshot_time, azure_tenant=tenant_id, rid=rid, data=data)
+
+	@staticmethod
+	def snapshot_time() -> datetime.datetime:
+		return datetime.datetime.now(datetime.timezone.utc)