Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[feat] Handle CloudFormation notifications #178

Merged
merged 3 commits into from
Nov 22, 2023
Merged

[feat] Handle CloudFormation notifications #178

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c42743c
[feat] Handle CloudFormation notifications
aquamatthias Nov 22, 2023
097c218
add context
aquamatthias Nov 22, 2023
8d51298
also handle create/update/delete
aquamatthias Nov 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 57 additions & 14 deletions fixbackend/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,17 @@
from dataclasses import replace
from datetime import timedelta
from ssl import Purpose, create_default_context
from typing import Any, AsyncIterator, Awaitable, Callable, ClassVar, Optional, Set, Tuple, cast
from typing import (
Any,
AsyncIterator,
Awaitable,
Callable,
ClassVar,
Optional,
Set,
Tuple,
cast,
)

import boto3
import httpx
Expand Down Expand Up @@ -46,7 +56,10 @@
from fixbackend.certificates.cert_store import CertificateStore
from fixbackend.cloud_accounts.account_setup import AwsAccountSetupHelper
from fixbackend.cloud_accounts.repository import CloudAccountRepositoryImpl
from fixbackend.cloud_accounts.router import cloud_accounts_callback_router, cloud_accounts_router
from fixbackend.cloud_accounts.router import (
cloud_accounts_callback_router,
cloud_accounts_router,
)
from fixbackend.cloud_accounts.service_impl import CloudAccountServiceImpl
from fixbackend.collect.collect_queue import RedisCollectQueue
from fixbackend.config import Config
Expand All @@ -63,7 +76,11 @@
from fixbackend.inventory.inventory_client import InventoryClient
from fixbackend.inventory.inventory_service import InventoryService
from fixbackend.inventory.router import inventory_router
from fixbackend.logging_context import get_logging_context, set_fix_cloud_account_id, set_workspace_id
from fixbackend.logging_context import (
get_logging_context,
set_fix_cloud_account_id,
set_workspace_id,
)
from fixbackend.metering.metering_repository import MeteringRepository
from fixbackend.middleware.x_real_ip import RealIpMiddleware
from fixbackend.subscription.aws_marketplace import AwsMarketplaceHandler
Expand All @@ -88,6 +105,7 @@ def fast_api_app(cfg: Config) -> FastAPI:
client_context = create_default_context(purpose=Purpose.SERVER_AUTH)
if ca_cert_path:
client_context.load_verify_locations(ca_cert_path)
http_client = deps.add(SN.http_client, AsyncClient(verify=ca_cert_path or True))

def create_redis(url: str) -> Redis:
kwargs = dict(ssl_ca_certs=ca_cert_path) if url.startswith("rediss://") else {}
Expand All @@ -97,7 +115,6 @@ def create_redis(url: str) -> Redis:

@asynccontextmanager
async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
http_client = deps.add(SN.http_client, AsyncClient(verify=ca_cert_path or True))
arq_redis = deps.add(
SN.arq_redis,
await create_pool(
Expand All @@ -111,7 +128,8 @@ async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
deps.add(SN.readonly_redis, create_redis(cfg.redis_readonly_url))
readwrite_redis = deps.add(SN.readwrite_redis, create_redis(cfg.redis_readwrite_url))
domain_event_subscriber = deps.add(
SN.domain_event_subscriber, DomainEventSubscriber(readwrite_redis, cfg, "fixbackend")
SN.domain_event_subscriber,
DomainEventSubscriber(readwrite_redis, cfg, "fixbackend"),
)
engine = deps.add(
SN.async_engine,
Expand All @@ -130,7 +148,10 @@ async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
deps.add(SN.collect_queue, RedisCollectQueue(arq_redis))
graph_db_access = deps.add(SN.graph_db_access, GraphDatabaseAccessManager(cfg, session_maker))
inventory_client = deps.add(SN.inventory_client, InventoryClient(cfg.inventory_url, http_client))
deps.add(SN.inventory, InventoryService(inventory_client, graph_db_access, domain_event_subscriber))
deps.add(
SN.inventory,
InventoryService(inventory_client, graph_db_access, domain_event_subscriber),
)
fixbackend_events = deps.add(
SN.domain_event_redis_stream_publisher,
RedisStreamPublisher(
Expand All @@ -142,7 +163,8 @@ async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
)
domain_event_publisher = deps.add(SN.domain_event_sender, DomainEventPublisherImpl(fixbackend_events))
workspace_repo = deps.add(
SN.workspace_repo, WorkspaceRepositoryImpl(session_maker, graph_db_access, domain_event_publisher)
SN.workspace_repo,
WorkspaceRepositoryImpl(session_maker, graph_db_access, domain_event_publisher),
)
subscription_repo = deps.add(SN.subscription_repo, SubscriptionRepository(session_maker))
deps.add(
Expand All @@ -160,7 +182,9 @@ async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
CustomerIoEventConsumer(http_client, cfg, domain_event_subscriber),
)
cloud_accounts_redis_publisher = RedisPubSubPublisher(
redis=readwrite_redis, channel="cloud_accounts", publisher_name="cloud_account_service"
redis=readwrite_redis,
channel="cloud_accounts",
publisher_name="cloud_account_service",
)
deps.add(
SN.cloud_account_service,
Expand All @@ -173,6 +197,9 @@ async def setup_teardown_application(_: FastAPI) -> AsyncIterator[None]:
cfg,
AwsAccountSetupHelper(boto_session),
dispatching=False,
http_client=http_client,
boto_session=boto_session,
cf_stack_queue_url=cfg.aws_cf_stack_notification_sqs_url,
),
)

Expand All @@ -199,7 +226,8 @@ async def setup_teardown_dispatcher(_: FastAPI) -> AsyncIterator[None]:
)
rw_redis = deps.add(SN.readwrite_redis, create_redis(cfg.redis_readwrite_url))
domain_event_subscriber = deps.add(
SN.domain_event_subscriber, DomainEventSubscriber(rw_redis, cfg, "dispatching")
SN.domain_event_subscriber,
DomainEventSubscriber(rw_redis, cfg, "dispatching"),
)
temp_store_redis = deps.add(SN.temp_store_redis, create_redis(cfg.redis_temp_store_url))
engine = deps.add(
Expand Down Expand Up @@ -231,10 +259,13 @@ async def setup_teardown_dispatcher(_: FastAPI) -> AsyncIterator[None]:

domain_event_publisher = deps.add(SN.domain_event_sender, DomainEventPublisherImpl(fixbackend_events))
workspace_repo = deps.add(
SN.workspace_repo, WorkspaceRepositoryImpl(session_maker, db_access, domain_event_publisher)
SN.workspace_repo,
WorkspaceRepositoryImpl(session_maker, db_access, domain_event_publisher),
)
cloud_accounts_redis_publisher = RedisPubSubPublisher(
redis=rw_redis, channel="cloud_accounts", publisher_name="cloud_account_service"
redis=rw_redis,
channel="cloud_accounts",
publisher_name="cloud_account_service",
)
deps.add(
SN.cloud_account_service,
Expand All @@ -247,6 +278,9 @@ async def setup_teardown_dispatcher(_: FastAPI) -> AsyncIterator[None]:
cfg,
AwsAccountSetupHelper(boto_session),
dispatching=True,
http_client=http_client,
boto_session=boto_session,
cf_stack_queue_url=cfg.aws_cf_stack_notification_sqs_url,
),
)
deps.add(
Expand Down Expand Up @@ -297,7 +331,8 @@ async def setup_teardown_billing(_: FastAPI) -> AsyncIterator[None]:
domain_event_publisher = deps.add(SN.domain_event_sender, DomainEventPublisherImpl(fixbackend_events))
metering_repo = deps.add(SN.metering_repo, MeteringRepository(session_maker))
workspace_repo = deps.add(
SN.workspace_repo, WorkspaceRepositoryImpl(session_maker, graph_db_access, domain_event_publisher)
SN.workspace_repo,
WorkspaceRepositoryImpl(session_maker, graph_db_access, domain_event_publisher),
)
subscription_repo = deps.add(SN.subscription_repo, SubscriptionRepository(session_maker))
aws_marketplace = deps.add(
Expand Down Expand Up @@ -434,7 +469,11 @@ async def refresh_session(request: Request, call_next: Callable[[Request], Await
return response

if cfg.static_assets:
app.mount("/", StaticFiles(directory=cfg.static_assets, html=True), name="static_assets")
app.mount(
"/",
StaticFiles(directory=cfg.static_assets, html=True),
name="static_assets",
)

@app.get("/")
async def root(request: Request) -> Response:
Expand Down Expand Up @@ -466,7 +505,11 @@ def setup_process() -> FastAPI:
"""
current_config = config.get_config()
level = logging.DEBUG if current_config.args.debug else logging.INFO
setup_logger(f"fixbackend_{current_config.args.mode}", level=level, get_logging_context=get_logging_context)
setup_logger(
f"fixbackend_{current_config.args.mode}",
level=level,
get_logging_context=get_logging_context,
)

# Replace all special uvicorn handlers
for logger in ["uvicorn", "uvicorn.error", "uvicorn.access"]:
Expand Down
12 changes: 12 additions & 0 deletions fixbackend/cloud_accounts/models/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ class GcpCloudAccess(CloudAccess):
class CloudAccountState(ABC):
state_name: ClassVar[str]

def cloud_access(self) -> Optional[CloudAccess]:
return None


class CloudAccountStates:
"""
Expand Down Expand Up @@ -85,6 +88,9 @@ class Discovered(CloudAccountState):
state_name: ClassVar[str] = "discovered"
access: CloudAccess

def cloud_access(self) -> Optional[CloudAccess]:
return self.access

@frozen
class Configured(CloudAccountState):
"""
Expand All @@ -95,6 +101,9 @@ class Configured(CloudAccountState):
access: CloudAccess
enabled: bool # is enabled for collection

def cloud_access(self) -> Optional[CloudAccess]:
return self.access

@frozen
class Degraded(CloudAccountState):
"""
Expand All @@ -105,6 +114,9 @@ class Degraded(CloudAccountState):
access: CloudAccess
error: str

def cloud_access(self) -> Optional[CloudAccess]:
return self.access


@frozen(kw_only=True)
class CloudAccount:
Expand Down
Loading