pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

provision portfolio state machine

Browse Source
This commit is contained in:
Philip Kalinsky
2020-01-08 11:01:55 -05:00
committed by tomdds
parent ad82706bd4
commit 69bd2f43a5
22 changed files with 1122 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
atst/domain/csp/__init__.py
View File

@@ -1,3 +1,5 @@
import importlib
from .cloud import MockCloudProvider
from .file_uploads import AzureUploader, MockUploader
from .reports import MockReportingProvider
@@ -29,3 +31,19 @@ def make_csp_provider(app, csp=None):
app.csp = MockCSP(app, test_mode=True)
else:
app.csp = MockCSP(app)
def _stage_to_classname(stage):
return "".join(map(lambda word: word.capitalize(), stage.replace('_', ' ').split(" ")))
def get_stage_csp_class(stage, class_type):
"""
given a stage name and class_type return the class
class_type is either 'payload' or 'result'
"""
cls_name = "".join([_stage_to_classname(stage), "CSP", class_type.capitalize()])
try:
return getattr(importlib.import_module("atst.domain.csp.cloud"), cls_name)
except AttributeError:
print("could not import CSP Result class <%s>" % cls_name)

459
atst/domain/csp/cloud.py
View File

@@ -1,12 +1,12 @@
from typing import Dict
import re
from typing import Dict
from uuid import uuid4
from pydantic import BaseModel
from atst.models.user import User
from atst.models.application import Application
from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole
from .policy import AzurePolicyManager
class GeneralCSPException(Exception):
@@ -142,6 +142,92 @@ class BaselineProvisionException(GeneralCSPException):
)
class BaseCSPPayload(BaseModel):
#{"username": "mock-cloud", "pass": "shh"}
creds: Dict
class TenantCSPPayload(BaseCSPPayload):
user_id: str
password: str
domain_name: str
first_name: str
last_name: str
country_code: str
password_recovery_email_address: str
class TenantCSPResult(BaseModel):
user_id: str
tenant_id: str
user_object_id: str
class BillingProfileAddress(BaseModel):
address: Dict
"""
"address": {
"firstName": "string",
"lastName": "string",
"companyName": "string",
"addressLine1": "string",
"addressLine2": "string",
"addressLine3": "string",
"city": "string",
"region": "string",
"country": "string",
"postalCode": "string"
},
"""
class BillingProfileCLINBudget(BaseModel):
clinBudget: Dict
"""
"clinBudget": {
"amount": 0,
"startDate": "2019-12-18T16:47:40.909Z",
"endDate": "2019-12-18T16:47:40.909Z",
"externalReferenceId": "string"
}
"""
class BillingProfileCSPPayload(BaseCSPPayload, BillingProfileAddress, BillingProfileCLINBudget):
displayName: str
poNumber: str
invoiceEmailOptIn: str
"""
{
"displayName": "string",
"poNumber": "string",
"address": {
"firstName": "string",
"lastName": "string",
"companyName": "string",
"addressLine1": "string",
"addressLine2": "string",
"addressLine3": "string",
"city": "string",
"region": "string",
"country": "string",
"postalCode": "string"
},
"invoiceEmailOptIn": true,
Note: These last 2 are also the body for adding/updating new TOs/clins
"enabledAzurePlans": [
{
"skuId": "string"
}
],
"clinBudget": {
"amount": 0,
"startDate": "2019-12-18T16:47:40.909Z",
"endDate": "2019-12-18T16:47:40.909Z",
"externalReferenceId": "string"
}
}
"""
class CloudProviderInterface:
def root_creds(self) -> Dict:
raise NotImplementedError()
@@ -325,6 +411,71 @@ class MockCloudProvider(CloudProviderInterface):
return {"id": self._id(), "credentials": self._auth_credentials}
def create_tenant(self, payload):
"""
payload is an instance of TenantCSPPayload data class
"""
self._authorize(payload.creds)
self._delay(1, 5)
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
# return tenant id, tenant owner id and tenant owner object id from:
response = {"tenantId": "string", "userId": "string", "objectId": "string"}
return {
"tenant_id": response["tenantId"],
"user_id": response["userId"],
"user_object_id": response["objectId"],
}
def create_billing_profile(self, creds, tenant_admin_details, billing_owner_id):
# call billing profile creation endpoint, specifying owner
# Payload:
"""
{
"displayName": "string",
"poNumber": "string",
"address": {
"firstName": "string",
"lastName": "string",
"companyName": "string",
"addressLine1": "string",
"addressLine2": "string",
"addressLine3": "string",
"city": "string",
"region": "string",
"country": "string",
"postalCode": "string"
},
"invoiceEmailOptIn": true,
Note: These last 2 are also the body for adding/updating new TOs/clins
"enabledAzurePlans": [
{
"skuId": "string"
}
],
"clinBudget": {
"amount": 0,
"startDate": "2019-12-18T16:47:40.909Z",
"endDate": "2019-12-18T16:47:40.909Z",
"externalReferenceId": "string"
}
}
"""
# response will be mostly the same as the body, but we only really care about the id
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
response = {"id": "string"}
return {"billing_profile_id": response["id"]}
def create_or_update_user(self, auth_credentials, user_info, csp_role_id):
self._authorize(auth_credentials)
@@ -401,18 +552,15 @@ REMOTE_ROOT_ROLE_DEF_ID = "/providers/Microsoft.Authorization/roleDefinitions/00
class AzureSDKProvider(object):
def __init__(self):
from azure.mgmt import subscription, authorization, managementgroups
from azure.mgmt.resource import policy
from azure.mgmt import subscription, authorization
import azure.graphrbac as graphrbac
import azure.common.credentials as credentials
from msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD
self.subscription = subscription
self.authorization = authorization
self.managementgroups = managementgroups
self.graphrbac = graphrbac
self.credentials = credentials
self.policy = policy
# may change to a JEDI cloud
self.cloud = AZURE_PUBLIC_CLOUD
@@ -430,28 +578,45 @@ class AzureCloudProvider(CloudProviderInterface):
else:
self.sdk = azure_sdk_provider
self.policy_manager = AzurePolicyManager(config["AZURE_POLICY_LOCATION"])
def create_environment(
self, auth_credentials: Dict, user: User, environment: Environment
):
# since this operation would only occur within a tenant, should we source the tenant
# via lookup from environment once we've created the portfolio csp data schema
# something like this:
# environment_tenant = environment.application.portfolio.csp_data.get('tenant_id', None)
# though we'd probably source the whole credentials for these calls from the portfolio csp
# data, as it would have to be where we store the creds for the at-at user within the portfolio tenant
# credentials = self._get_credential_obj(environment.application.portfolio.csp_data.get_creds())
credentials = self._get_credential_obj(self._root_creds)
display_name = f"{environment.application.name}_{environment.name}_{environment.id}" # proposed format
management_group_id = "?" # management group id chained from environment
parent_id = "?" # from environment.application
sub_client = self.sdk.subscription.SubscriptionClient(credentials)
management_group = self._create_management_group(
credentials, management_group_id, display_name, parent_id,
display_name = f"{environment.application.name}_{environment.name}_{environment.id}" # proposed format
billing_profile_id = "?" # something chained from environment?
sku_id = AZURE_SKU_ID
# we want to set AT-AT as an owner here
# we could potentially associate subscriptions with "management groups" per DOD component
body = self.sdk.subscription.models.ModernSubscriptionCreationParameters(
display_name,
billing_profile_id,
sku_id,
# owner=<AdPrincipal: for AT-AT user>
)
return management_group
# These 2 seem like something that might be worthwhile to allow tiebacks to
# TOs filed for the environment
billing_account_name = "?"
invoice_section_name = "?"
# We may also want to create billing sections in the enrollment account
sub_creation_operation = sub_client.subscription_factory.create_subscription(
billing_account_name, invoice_section_name, body
)
# the resulting object from this process is a link to the new subscription
# not a subscription model, so we'll have to unpack the ID
new_sub = sub_creation_operation.result()
subscription_id = self._extract_subscription_id(new_sub.subscription_link)
if subscription_id:
return subscription_id
else:
# troublesome error, subscription should exist at this point
# but we just don't have a valid ID
pass
def create_atat_admin_user(
self, auth_credentials: Dict, csp_environment_id: str
@@ -490,126 +655,136 @@ class AzureCloudProvider(CloudProviderInterface):
"role_name": role_assignment_id,
}
def _create_application(self, auth_credentials: Dict, application: Application):
management_group_name = str(uuid4()) # can be anything, not just uuid
display_name = application.name # Does this need to be unique?
credentials = self._get_credential_obj(auth_credentials)
parent_id = "?" # application.portfolio.csp_details.management_group_id
return self._create_management_group(
credentials, management_group_name, display_name, parent_id,
def create_tenant(self, payload):
# auth as SP that is allowed to create tenant? (tenant creation sp creds)
# create tenant with owner details (populated from portfolio point of contact, pw is generated)
# return tenant id, tenant owner id and tenant owner object id from:
response = {"tenantId": "string", "userId": "string", "objectId": "string"}
return self._ok(
{
"tenant_id": response["tenantId"],
"user_id": response["userId"],
"user_object_id": response["objectId"],
}
)
def _create_management_group(
self, credentials, management_group_id, display_name, parent_id=None,
):
mgmgt_group_client = self.sdk.managementgroups.ManagementGroupsAPI(credentials)
create_parent_grp_info = self.sdk.managementgroups.models.CreateParentGroupInfo(
id=parent_id
)
create_mgmt_grp_details = self.sdk.managementgroups.models.CreateManagementGroupDetails(
parent=create_parent_grp_info
)
mgmt_grp_create = self.sdk.managementgroups.models.CreateManagementGroupRequest(
name=management_group_id,
display_name=display_name,
details=create_mgmt_grp_details,
)
create_request = mgmgt_group_client.management_groups.create_or_update(
management_group_id, mgmt_grp_create
)
def create_billing_owner(self, creds, tenant_admin_details):
# authenticate as tenant_admin
# create billing owner identity
# result is a synchronous wait, might need to do a poll instead to handle first mgmt group create
# since we were told it could take 10+ minutes to complete, unless this handles that polling internally
return create_request.result()
# TODO: Lookup response format
# Managed service identity?
response = {"id": "string"}
return self._ok({"billing_owner_id": response["id"]})
def _create_subscription(
self,
credentials,
display_name,
billing_profile_id,
sku_id,
management_group_id,
billing_account_name,
invoice_section_name,
):
sub_client = self.sdk.subscription.SubscriptionClient(credentials)
billing_profile_id = "?" # where do we source this?
sku_id = AZURE_SKU_ID
# These 2 seem like something that might be worthwhile to allow tiebacks to
# TOs filed for the environment
billing_account_name = "?" # from TO?
invoice_section_name = "?" # from TO?
body = self.sdk.subscription.models.ModernSubscriptionCreationParameters(
display_name=display_name,
billing_profile_id=billing_profile_id,
sku_id=sku_id,
management_group_id=management_group_id,
)
# We may also want to create billing sections in the enrollment account
sub_creation_operation = sub_client.subscription_factory.create_subscription(
billing_account_name, invoice_section_name, body
)
# the resulting object from this process is a link to the new subscription
# not a subscription model, so we'll have to unpack the ID
new_sub = sub_creation_operation.result()
subscription_id = self._extract_subscription_id(new_sub.subscription_link)
if subscription_id:
return subscription_id
else:
# troublesome error, subscription should exist at this point
# but we just don't have a valid ID
pass
AZURE_MANAGEMENT_API = "https://management.azure.com"
def _create_policy_definition(
self, credentials, subscription_id, management_group_id, properties,
):
def assign_billing_owner(self, creds, billing_owner_id, tenant_id):
# TODO: Do we source role definition ID from config, api or self-defined?
# TODO: If from api,
"""
Requires credentials that have AZURE_MANAGEMENT_API
specified as the resource. The Service Principal
specified in the credentials must have the "Resource
Policy Contributor" role assigned with a scope at least
as high as the management group specified by
management_group_id.
Arguments:
credentials -- ServicePrincipalCredentials
subscription_id -- str, ID of the subscription (just the UUID, not the path)
management_group_id -- str, ID of the management group (just the UUID, not the path)
properties -- dictionary, the "properties" section of a valid Azure policy definition document
Returns:
azure.mgmt.resource.policy.[api version].models.PolicyDefinition: the PolicyDefinition object provided to Azure
Raises:
TBD
{
"principalId": "string",
"principalTenantId": "string",
"billingRoleDefinitionId": "string"
}
"""
# TODO: which subscription would this be?
client = self.sdk.policy.PolicyClient(credentials, subscription_id)
definition = client.policy_definitions.models.PolicyDefinition(
policy_type=properties.get("policyType"),
mode=properties.get("mode"),
display_name=properties.get("displayName"),
description=properties.get("description"),
policy_rule=properties.get("policyRule"),
parameters=properties.get("parameters"),
return self.ok()
def create_billing_profile(self, creds, tenant_admin_details, billing_owner_id):
# call billing profile creation endpoint, specifying owner
# Payload:
"""
{
"displayName": "string",
"poNumber": "string",
"address": {
"firstName": "string",
"lastName": "string",
"companyName": "string",
"addressLine1": "string",
"addressLine2": "string",
"addressLine3": "string",
"city": "string",
"region": "string",
"country": "string",
"postalCode": "string"
},
"invoiceEmailOptIn": true,
Note: These last 2 are also the body for adding/updating new TOs/clins
"enabledAzurePlans": [
{
"skuId": "string"
}
],
"clinBudget": {
"amount": 0,
"startDate": "2019-12-18T16:47:40.909Z",
"endDate": "2019-12-18T16:47:40.909Z",
"externalReferenceId": "string"
}
}
"""
# response will be mostly the same as the body, but we only really care about the id
response = {"id": "string"}
return self._ok({"billing_profile_id": response["id"]})
def report_clin(self, creds, clin_id, clin_amount, clin_start, clin_end, clin_to):
# should consumer be responsible for reporting each clin or
# should this take a list and manage the sequential reporting?
""" Payload
{
"enabledAzurePlans": [
{
"skuId": "string"
}
],
"clinBudget": {
"amount": 0,
"startDate": "2019-12-18T16:47:40.909Z",
"endDate": "2019-12-18T16:47:40.909Z",
"externalReferenceId": "string"
}
}
"""
# we don't need any of the returned info for this
return self._ok()
def create_remote_admin(self, creds, tenant_details):
# create app/service principal within tenant, with name constructed from tenant details
# assign principal global admin
# needs to call out to CLI with tenant owner username/password, prototyping for that underway
# return identifier and creds to consumer for storage
response = {"clientId": "string", "secretKey": "string", "tenantId": "string"}
return self._ok(
{
"client_id": response["clientId"],
"secret_key": response["secret_key"],
"tenant_id": response["tenantId"],
}
)
name = properties.get("displayName")
def force_tenant_admin_pw_update(self, creds, tenant_owner_id):
# use creds to update to force password recovery?
# not sure what the endpoint/method for this is, yet
return client.policy_definitions.create_or_update_at_management_group(
policy_definition_name=name,
parameters=definition,
management_group_id=management_group_id,
)
return self._ok()
def create_billing_alerts(self, TBD):
# TODO: Add azure-mgmt-consumption for Budget and Notification entities/operations
# TODO: Determine how to auth against that API using the SDK, doesn't seeem possible at the moment
# TODO: billing alerts are registered as Notifications on Budget objects, which have start/end dates
# TODO: determine what the keys in the Notifications dict are supposed to be
# we may need to rotate budget objects when new TOs/CLINs are reported?
# we likely only want the budget ID, can be updated or replaced?
response = {"id": "id"}
return self._ok({"budget_id": response["id"]})
def _get_management_service_principal(self):
# we really should be using graph.microsoft.com, but i'm getting
@@ -663,6 +838,7 @@ class AzureCloudProvider(CloudProviderInterface):
return sub_id_match.group(1)
def _get_credential_obj(self, creds, resource=None):
return self.sdk.credentials.ServicePrincipalCredentials(
client_id=creds.get("client_id"),
secret=creds.get("secret_key"),
@@ -671,6 +847,27 @@ class AzureCloudProvider(CloudProviderInterface):
cloud_environment=self.sdk.cloud,
)
def _make_tenant_admin_cred_obj(self, username, password):
return self.sdk.credentials.UserPassCredentials(username, password)
def _ok(self, body=None):
return self._make_response("ok", body)
def _error(self, body=None):
return self._make_response("error", body)
def _make_response(self, status, body=dict()):
"""Create body for responses from API
Arguments:
status {string} -- "ok" or "error"
body {dict} -- dict containing details of response or error, if applicable
Returns:
dict -- status of call with body containing details
"""
return {"status": status, "body": body}
@property
def _root_creds(self):
return {

1
atst/domain/portfolios/__init__.py
View File

@@ -2,4 +2,5 @@ from .portfolios import (
Portfolios,
PortfolioError,
PortfolioDeletionApplicationsExistError,
PortfolioStateMachines,
)

64
atst/domain/portfolios/portfolios.py
View File

@@ -1,11 +1,15 @@
from typing import List
from uuid import UUID
from atst.database import db
from atst.domain.permission_sets import PermissionSets
from atst.domain.authz import Authorization
from atst.domain.portfolio_roles import PortfolioRoles
from atst.domain.invitations import PortfolioInvitations
from atst.models import Permissions, PortfolioRole, PortfolioRoleStatus
from .query import PortfoliosQuery
from atst.domain.invitations import PortfolioInvitations
from atst.models import Portfolio, PortfolioStateMachine, FSMStates, Permissions, PortfolioRole, PortfolioRoleStatus
from .query import PortfoliosQuery, PortfolioStateMachinesQuery
from .scopes import ScopedPortfolio
@@ -17,7 +21,24 @@ class PortfolioDeletionApplicationsExistError(Exception):
pass
class PortfolioStateMachines(object):
@classmethod
def create(cls, portfolio, **sm_attrs):
sm_attrs.update({'portfolio': portfolio})
sm = PortfolioStateMachinesQuery.create(**sm_attrs)
return sm
class Portfolios(object):
@classmethod
def get_or_create_state_machine(cls, portfolio):
"""
get or create Portfolio State Machine for a Portfolio
"""
return portfolio.state_machine or PortfolioStateMachines.create(portfolio)
@classmethod
def create(cls, user, portfolio_attrs):
portfolio = PortfoliosQuery.create(**portfolio_attrs)
@@ -111,3 +132,40 @@ class Portfolios(object):
portfolio.description = new_data["description"]
PortfoliosQuery.add_and_commit(portfolio)
@classmethod
def base_provision_query(cls):
return (
db.session.query(Portfolio.id)
)
@classmethod
def get_portfolios_pending_provisioning(cls) -> List[UUID]:
"""
Any portfolio with a corresponding State Machine that is either:
not started yet,
failed in creating a tenant
failed
"""
results = (
cls.base_provision_query().\
join(PortfolioStateMachine).\
filter(
or_(
PortfolioStateMachine.state == FSMStates.UNSTARTED,
PortfolioStateMachine.state == FSMStates.FAILED,
PortfolioStateMachine.state == FSMStates.TENANT_CREATION_FAILED,
)
)
)
return [id_ for id_, in results]
#db.session.query(PortfolioStateMachine).\
# filter(
# or_(
# PortfolioStateMachine.state==FSMStates.UNSTARTED,
# PortfolioStateMachine.state==FSMStates.UNSTARTED,
# )
# ).all()

6
atst/domain/portfolios/query.py
View File

@@ -8,6 +8,12 @@ from atst.models.application_role import (
Status as ApplicationRoleStatus,
)
from atst.models.application import Application
from atst.models.portfolio_state_machine import PortfolioStateMachine
#from atst.models.application import Application
class PortfolioStateMachinesQuery(Query):
model = PortfolioStateMachine
class PortfoliosQuery(Query):

34
atst/jobs.py
View File

@@ -7,14 +7,28 @@ from atst.models import (
EnvironmentJobFailure,
EnvironmentRoleJobFailure,
EnvironmentRole,
PortfolioJobFailure,
FSMStates,
)
from atst.domain.csp.cloud import CloudProviderInterface, GeneralCSPException
from atst.domain.environments import Environments
from atst.domain.portfolios import Portfolios
from atst.domain.portfolios.query import PortfolioStateMachinesQuery
from atst.domain.environment_roles import EnvironmentRoles
from atst.models.utils import claim_for_update
from atst.utils.localization import translate
class RecordPortfolioFailure(celery.Task):
def on_failure(self, exc, task_id, args, kwargs, einfo):
if "portfolio_id" in kwargs:
failure = PortfolioJobFailure(
portfolio_id=kwargs["portfolio_id"], task_id=task_id
)
db.session.add(failure)
db.session.commit()
class RecordEnvironmentFailure(celery.Task):
def on_failure(self, exc, task_id, args, kwargs, einfo):
if "environment_id" in kwargs:
@@ -50,6 +64,7 @@ def send_notification_mail(recipients, subject, body):
app.mailer.send(recipients, subject, body)
def do_create_environment(csp: CloudProviderInterface, environment_id=None):
environment = Environments.get(environment_id)
@@ -125,6 +140,16 @@ def do_work(fn, task, csp, **kwargs):
raise task.retry(exc=e)
def do_provision_portfolio(csp: CloudProviderInterface, portfolio_id=None):
portfolio = Portfolios.get_for_update(portfolio_id)
fsm = Portfolios.get_or_create_state_machine(portfolio)
fsm.trigger_next_transition()
@celery.task(bind=True, base=RecordPortfolioFailure)
def provision_portfolio(self, portfolio_id=None):
do_work(do_provision_portfolio, self, app.csp.cloud, portfolio_id=portfolio_id)
@celery.task(bind=True, base=RecordEnvironmentFailure)
def create_environment(self, environment_id=None):
do_work(do_create_environment, self, app.csp.cloud, environment_id=environment_id)
@@ -144,6 +169,15 @@ def provision_user(self, environment_role_id=None):
)
@celery.task(bind=True)
def dispatch_provision_portfolio(self):
"""
Iterate over portfolios with a corresponding State Machine that have not completed.
"""
for portfolio_id in Portfolios.get_portfolios_pending_provisioning():
provision_portfolio.delay(portfolio_id=portfolio_id)
@celery.task(bind=True)
def dispatch_create_environment(self):
for environment_id in Environments.get_environments_pending_creation(

3
atst/models/__init__.py
View File

@@ -7,11 +7,12 @@ from .audit_event import AuditEvent
from .clin import CLIN, JEDICLINType
from .environment import Environment
from .environment_role import EnvironmentRole, CSPRole
from .job_failure import EnvironmentJobFailure, EnvironmentRoleJobFailure
from .job_failure import EnvironmentJobFailure, EnvironmentRoleJobFailure, PortfolioJobFailure
from .notification_recipient import NotificationRecipient
from .permissions import Permissions
from .permission_set import PermissionSet
from .portfolio import Portfolio
from .portfolio_state_machine import PortfolioStateMachine, FSMStates
from .portfolio_invitation import PortfolioInvitation
from .portfolio_role import PortfolioRole, Status as PortfolioRoleStatus
from .task_order import TaskOrder

6
atst/models/job_failure.py
View File

@@ -14,3 +14,9 @@ class EnvironmentRoleJobFailure(Base, mixins.JobFailureMixin):
__tablename__ = "environment_role_job_failures"
environment_role_id = Column(ForeignKey("environment_roles.id"), nullable=False)
class PortfolioJobFailure(Base, mixins.JobFailureMixin):
__tablename__ = "portfolio_job_failures"
portfolio_id = Column(ForeignKey("portfolios.id"), nullable=False)

1
atst/models/mixins/__init__.py
View File

@@ -4,3 +4,4 @@ from .permissions import PermissionsMixin
from .deletable import DeletableMixin
from .invites import InvitesMixin
from .job_failure import JobFailureMixin
from .state_machines import FSMMixin

109
atst/models/mixins/state_machines.py Normal file
View File

@@ -0,0 +1,109 @@
from enum import Enum
from atst.database import db
class StageStates(Enum):
CREATED = "created"
IN_PROGRESS = "in progress"
FAILED = "failed"
class AzureStages(Enum):
TENANT = "tenant"
BILLING_PROFILE = "billing profile"
ADMIN_SUBSCRIPTION = "admin subscription"
def _build_csp_states(csp_stages):
states = {
'UNSTARTED' : "unstarted",
'STARTING' : "starting",
'STARTED' : "started",
'COMPLETED' : "completed",
'FAILED' : "failed",
}
for csp_stage in csp_stages:
for state in StageStates:
states[csp_stage.name+"_"+state.name] = csp_stage.value+" "+state.value
return states
FSMStates = Enum('FSMStates', _build_csp_states(AzureStages))
def _build_transitions(csp_stages):
transitions = []
states = []
compose_state = lambda csp_stage, state: getattr(FSMStates, "_".join([csp_stage.name, state.name]))
for stage_i, csp_stage in enumerate(csp_stages):
for state in StageStates:
states.append(dict(name=compose_state(csp_stage, state), tags=[csp_stage.name, state.name]))
if state == StageStates.CREATED:
if stage_i > 0:
src = compose_state(list(csp_stages)[stage_i-1] , StageStates.CREATED)
else:
src = FSMStates.STARTED
transitions.append(
dict(
trigger='create_'+csp_stage.name.lower(),
source=src,
dest=compose_state(csp_stage, StageStates.IN_PROGRESS),
after='after_in_progress_callback',
)
)
if state == StageStates.IN_PROGRESS:
transitions.append(
dict(
trigger='finish_'+csp_stage.name.lower(),
source=compose_state(csp_stage, state),
dest=compose_state(csp_stage, StageStates.CREATED),
conditions=['is_csp_data_valid'],
)
)
if state == StageStates.FAILED:
transitions.append(
dict(
trigger='fail_'+csp_stage.name.lower(),
source=compose_state(csp_stage, StageStates.IN_PROGRESS),
dest=compose_state(csp_stage, StageStates.FAILED),
)
)
return states, transitions
class FSMMixin():
system_states = [
{'name': FSMStates.UNSTARTED.name, 'tags': ['system']},
{'name': FSMStates.STARTING.name, 'tags': ['system']},
{'name': FSMStates.STARTED.name, 'tags': ['system']},
{'name': FSMStates.FAILED.name, 'tags': ['system']},
{'name': FSMStates.COMPLETED.name, 'tags': ['system']},
]
system_transitions = [
{'trigger': 'init', 'source': FSMStates.UNSTARTED, 'dest': FSMStates.STARTING},
{'trigger': 'start', 'source': FSMStates.STARTING, 'dest': FSMStates.STARTED},
{'trigger': 'reset', 'source': '*', 'dest': FSMStates.UNSTARTED},
{'trigger': 'fail', 'source': '*', 'dest': FSMStates.FAILED,}
]
def prepare_init(self, event): pass
def before_init(self, event): pass
def after_init(self, event): pass
def prepare_start(self, event): pass
def before_start(self, event): pass
def after_start(self, event): pass
def prepare_reset(self, event): pass
def before_reset(self, event): pass
def after_reset(self, event): pass
def fail_stage(self, stage):
fail_trigger = 'fail'+stage
if fail_trigger in self.machine.get_triggers(self.current_state.name):
self.trigger(fail_trigger)
def finish_stage(self, stage):
finish_trigger = 'finish_'+stage
if finish_trigger in self.machine.get_triggers(self.current_state.name):
self.trigger(finish_trigger)

23
atst/models/portfolio.py
View File

@@ -11,6 +11,9 @@ from atst.domain.permission_sets import PermissionSets
from atst.utils import first_or_none
from atst.database import db
from sqlalchemy_json import NestedMutableJson
class Portfolio(
Base, mixins.TimestampsMixin, mixins.AuditableMixin, mixins.DeletableMixin
@@ -19,16 +22,30 @@ class Portfolio(
id = types.Id()
name = Column(String, nullable=False)
description = Column(String)
defense_component = Column(
ARRAY(String), nullable=False
String, nullable=False
) # Department of Defense Component
app_migration = Column(String) # App Migration
complexity = Column(ARRAY(String)) # Application Complexity
complexity_other = Column(String)
description = Column(String)
dev_team = Column(ARRAY(String)) # Development Team
dev_team_other = Column(String)
native_apps = Column(String) # Native Apps
team_experience = Column(String) # Team Experience
csp_data = Column(NestedMutableJson, nullable=True)
applications = relationship(
"Application",
back_populates="portfolio",
primaryjoin="and_(Application.portfolio_id == Portfolio.id, Application.deleted == False)",
)
state_machine = relationship("PortfolioStateMachine",
uselist=False, back_populates="portfolio")
roles = relationship("PortfolioRole")
task_orders = relationship("TaskOrder")
@@ -77,7 +94,7 @@ class Portfolio(
"""
Return the earliest period of performance start date and latest period
of performance end date for all active task orders in a portfolio.
@return: (datetime.date or None, datetime.date or None)
@return: (datetime.date or None, datetime.date or None)
"""
start_dates = (
task_order.start_date

182
atst/models/portfolio_state_machine.py Normal file
View File

@@ -0,0 +1,182 @@
import importlib
from sqlalchemy import Column, ForeignKey, Enum as SQLAEnum
from sqlalchemy.orm import relationship, reconstructor
from sqlalchemy.dialects.postgresql import UUID
from pydantic import ValidationError as PydanticValidationError
from transitions import Machine
from transitions.extensions.states import add_state_features, Tags
from flask import current_app as app
from atst.domain.csp.cloud import ConnectionException, UnknownServerException
from atst.domain.csp import MockCSP, AzureCSP, get_stage_csp_class
from atst.database import db
from atst.queue import celery
from atst.models.types import Id
from atst.models.base import Base
import atst.models.mixins as mixins
from atst.models.mixins.state_machines import (
FSMStates, AzureStages, _build_transitions
)
@add_state_features(Tags)
class StateMachineWithTags(Machine):
pass
class PortfolioStateMachine(
Base, mixins.TimestampsMixin, mixins.AuditableMixin, mixins.DeletableMixin, mixins.FSMMixin,
):
__tablename__ = "portfolio_state_machines"
id = Id()
portfolio_id = Column(
UUID(as_uuid=True),
ForeignKey("portfolios.id"),
)
portfolio = relationship("Portfolio", back_populates="state_machine")
state = Column(
SQLAEnum(FSMStates, native_enum=False, create_constraint=False),
default=FSMStates.UNSTARTED, nullable=False
)
def __init__(self, portfolio, csp=None, **kwargs):
self.portfolio = portfolio
self.attach_machine()
def after_state_change(self, event):
db.session.add(self)
db.session.commit()
@reconstructor
def attach_machine(self):
"""
This is called as a result of a sqlalchemy query.
Attach a machine depending on the current state.
"""
self.machine = StateMachineWithTags(
model = self,
send_event=True,
initial=self.current_state if self.state else FSMStates.UNSTARTED,
auto_transitions=False,
after_state_change='after_state_change',
)
states, transitions = _build_transitions(AzureStages)
self.machine.add_states(self.system_states+states)
self.machine.add_transitions(self.system_transitions+transitions)
@property
def current_state(self):
if isinstance(self.state, str):
return getattr(FSMStates, self.state)
return self.state
def trigger_next_transition(self):
state_obj = self.machine.get_state(self.state)
if state_obj.is_system:
if self.current_state in (FSMStates.UNSTARTED, FSMStates.STARTING):
# call the first trigger availabe for these two system states
trigger_name = self.machine.get_triggers(self.current_state.name)[0]
self.trigger(trigger_name)
elif self.current_state == FSMStates.STARTED:
# get the first trigger that starts with 'create_'
create_trigger = list(filter(lambda trigger: trigger.startswith('create_'),
self.machine.get_triggers(FSMStates.STARTED.name)))[0]
self.trigger(create_trigger)
elif state_obj.is_IN_PROGRESS:
pass
#elif state_obj.is_TENANT:
# pass
#elif state_obj.is_BILLING_PROFILE:
# pass
#@with_payload
def after_in_progress_callback(self, event):
stage = self.current_state.name.split('_IN_PROGRESS')[0].lower()
if stage == 'tenant':
payload = dict(
creds={"username": "mock-cloud", "pass": "shh"},
user_id='123',
password='123',
domain_name='123',
first_name='john',
last_name='doe',
country_code='US',
password_recovery_email_address='password@email.com'
)
elif stage == 'billing_profile':
payload = dict(
creds={"username": "mock-cloud", "pass": "shh"},
)
payload_data_cls = get_stage_csp_class(stage, "payload")
if not payload_data_cls:
self.fail_stage(stage)
try:
payload_data = payload_data_cls(**payload)
except PydanticValidationError as exc:
print(exc.json())
self.fail_stage(stage)
csp = event.kwargs.get('csp')
if csp is not None:
self.csp = AzureCSP(app).cloud
else:
self.csp = MockCSP(app).cloud
for attempt in range(5):
try:
response = getattr(self.csp, 'create_'+stage)(payload_data)
except (ConnectionException, UnknownServerException) as exc:
print('caught exception. retry', attempt)
continue
else: break
else:
# failed all attempts
self.fail_stage(stage)
if self.portfolio.csp_data is None:
self.portfolio.csp_data = {}
self.portfolio.csp_data[stage+"_data"] = response
db.session.add(self.portfolio)
db.session.commit()
self.finish_stage(stage)
def is_csp_data_valid(self, event):
# check portfolio csp details json field for fields
if self.portfolio.csp_data is None or \
not isinstance(self.portfolio.csp_data, dict):
return False
stage = self.current_state.name.split('_IN_PROGRESS')[0].lower()
stage_data = self.portfolio.csp_data.get(stage+"_data")
cls = get_stage_csp_class(stage, "result")
if not cls:
return False
try:
cls(**stage_data)
except PydanticValidationError as exc:
print(exc.json())
return False
return True
#print('failed condition', self.portfolio.csp_data)
@property
def application_id(self):
return None

4
atst/queue.py
View File

@@ -7,6 +7,10 @@ celery = Celery(__name__)
def update_celery(celery, app):
celery.conf.update(app.config)
celery.conf.CELERYBEAT_SCHEDULE = {
"beat-dispatch_provision_portfolio": {
"task": "atst.jobs.dispatch_provision_portfolio",
"schedule": 60,
},
"beat-dispatch_create_environment": {
"task": "atst.jobs.dispatch_create_environment",
"schedule": 60,