Browse Source

Document staging api

The purpose of the document staging api is to provide a client of
shipyard to ingest and manipulate documents within the UCP platform

In support of the need to reach out to other services, this change
also introduces the use of keystone for service discovery and connection
to other UCP services.

Change-Id: I6fa113f8786cad2884c0b791788a4ef40cd1a6b6
changes/87/569187/1
Hassan Kaous 5 years ago committed by Bryan Strassner
parent
commit
3ac47328c3
  1. 1
      AUTHORS
  2. 22
      alembic/versions/51b92375e5c4_initial_shipyard_base.py
  3. 20
      docs/API.md
  4. 19
      etc/shipyard/policy.yaml.sample
  5. 73
      etc/shipyard/shipyard.conf.sample
  6. 134
      shipyard_airflow/conf/config.py
  7. 28
      shipyard_airflow/control/actions_api.py
  8. 7
      shipyard_airflow/control/api.py
  9. 217
      shipyard_airflow/control/api_lock.py
  10. 8
      shipyard_airflow/control/base.py
  11. 178
      shipyard_airflow/control/configdocs_api.py
  12. 609
      shipyard_airflow/control/configdocs_helper.py
  13. 517
      shipyard_airflow/control/deckhand_client.py
  14. 67
      shipyard_airflow/control/rendered_configdocs_api.py
  15. 126
      shipyard_airflow/control/service_endpoints.py
  16. 12
      shipyard_airflow/db/common_db.py
  17. 105
      shipyard_airflow/db/shipyard_db.py
  18. 38
      shipyard_airflow/policy.py
  19. 22
      tests/unit/control/fake_response.py
  20. 146
      tests/unit/control/test_configdocs_api.py
  21. 643
      tests/unit/control/test_configdocs_helper.py
  22. 52
      tests/unit/control/test_rendered_configdocs_api.py

1
AUTHORS

@ -2,6 +2,7 @@ Alan Meadows <alan.meadows@gmail.com>
Anthony Lin <anthony.jclin@gmail.com>
Bryan Strassner <bryan.strassner@gmail.com>
Felipe Monteiro <felipe.monteiro@att.com>
Hassan Kaous <hkyq8@mst.edu>
Mark Burnett <mark.m.burnett@gmail.com>
One-Fine-Day <vd789v@att.com>
Pete Birley <pete@port.direct>

22
alembic/versions/51b92375e5c4_initial_shipyard_base.py

@ -61,7 +61,7 @@ def upgrade():
'action_command_audit',
# ID (ULID) of the audit
sa.Column('id', types.String(26), primary_key=True),
# The ID of action this audit record
# The ID of the action for this audit record
sa.Column('action_id', types.String(26), nullable=False),
# The text indicating command invoked
sa.Column('command', sa.Text, nullable=False),
@ -73,6 +73,25 @@ def upgrade():
server_default=func.now()),
)
op.create_table(
'api_locks',
# ID (ULID) of the lock
sa.Column('id', types.String(26), primary_key=True),
# The category/type of the lock
sa.Column('lock_type', types.String(20), nullable=False),
# Timestamp of when the lock was acquired
sa.Column('datetime',
types.TIMESTAMP(timezone=True),
server_default=func.now(),
nullable=False),
# Expires
sa.Column('expires', types.Integer, nullable=False, default=60),
# A marker if the lock is released
sa.Column('released', types.Boolean, nullable=False, default=False),
sa.Column('user', types.String(64), nullable=False),
sa.Column('reference_id', types.String(36), nullable=False),
)
def downgrade():
"""
@ -81,3 +100,4 @@ def downgrade():
op.drop_table('actions')
op.drop_table('preflight_validation_failures')
op.drop_table('action_command_audit')
op.drop_table('api_locks')

20
docs/API.md

@ -90,6 +90,9 @@ indicates that the specified collection should be deleted when the Shipyard
Buffer is committed. If a POST to the commitconfigdocs is in progress, this
POST should be rejected with a 409 error.
Important:
The expected input type for this request is 'Content-Type: application/x-yaml'
##### Query Parameters
* bufferMode=append|replace|**rejectOnContents**
Indicates how the existing Shipyard Buffer should be handled. By default,
@ -121,6 +124,9 @@ Returns the source documents for a collection of documents
* version=committed|**buffer**
Return the documents for the version specified - buffer by default.
Important:
The output type for this request is 'Content-Type: application/x-yaml'
##### Responses
* 200 OK
If documents can be retrieved.
@ -141,6 +147,10 @@ consider collections in any way.
#### GET /v1.0/renderedconfigdocs
Returns the full set of configdocs in their rendered form.
Important:
The output type for this request is 'Content-Type: application/x-yaml'
##### Query Parameters
* version=committed|**buffer**
Return the documents for the version specified - buffer by default.
@ -180,7 +190,7 @@ a 400 response. With force=true, allows for the commit to succeed (with a 200
response) even if there are validation failures from downstream components. The
aggregate response of validation failures will be returned in this case, but
the invalid documents will still be moved from the Shipyard Buffer to the
Committed Documents.
Committed Documents.
##### Responses
* 200 OK
@ -188,7 +198,7 @@ If the validations are successful. Returns an "empty" structure as as response
indicating no errors. A 200 may also be returned if there are validation
failures, but the force=true query parameter was specified. In this case, the
response will contain the list of validations.
* 400 Bad Request
* 400 Bad Request
If the validations fail. Returns a populated response structure containing
the aggregation of the failed validations.
* 409 Conflict
@ -290,13 +300,13 @@ airflow.
##### Responses
* 201 Created
If the action is created successfully, and all preconditions to run the DAG
are successful. The response body is the action entity created.
are successful. The response body is the action entity created.
* 400 Bad Request
If the action name doesn't exist, or the input entity is otherwise malformed.
If the action name doesn't exist, or the input entity is otherwise malformed.
* 409 Conflict
For any failed pre-run validations. The response body is the action entity
created, with the failed validations. The DAG will not begin execution in this
case.
case.
##### Example
```

19
etc/shipyard/policy.yaml.sample

@ -7,7 +7,7 @@
# Create a workflow action
# POST /api/v1.0/actions
#"workflow_orchestrator:create_actions": "rule:admin_required"
#"workflow_orchestrator:create_action": "rule:admin_required"
# Retreive an action by its id
# GET /api/v1.0/actions/{action_id}
@ -25,3 +25,20 @@
# POST /api/v1.0/actions/{action_id}/control/{control_verb}
#"workflow_orchestrator:invoke_action_control": "rule:admin_required"
# Ingest configuration documents for the site design
# POST /api/v1.0/configdocs/{collection_id}
#"workflow_orchestrator:create_configdocs": "rule:admin_required"
# Retrieve a collection of configuration documents
# GET /api/v1.0/configdocs/{collection_id}
#"workflow_orchestrator:get_configdocs": "rule:admin_required"
# Move documents from the Shipyard buffer to the committed documents
# POST /api/v1.0/commitconfigdocs
#"workflow_orchestrator:commit_configdocs": "rule:admin_required"
# Retrieve the configuration documents rendered by Deckhand into a
# complete design
# GET /api/v1.0/renderedconfigdocs
#"workflow_orchestrator:get_renderedconfigdocs": "rule:admin_required"

73
etc/shipyard/shipyard.conf.sample

@ -7,11 +7,10 @@
# From shipyard_airflow
#
# FQDN for the armada service (string value)
#host = armada-int.ucp
# Port for the armada service (integer value)
#port = 8000
# The service type for the service playing the role of Armada. The specified
# type is used to perform the service lookup in the Keystone service catalog.
# (string value)
#service_type = armada
[base]
@ -42,11 +41,10 @@
# From shipyard_airflow
#
# FQDN for the deckhand service (string value)
#host = deckhand-int.ucp
# Port for the deckhand service (integer value)
#port = 80
# The service type for the service playing the role of Deckhand. The specified
# type is used to perform the service lookup in the Keystone service catalog.
# (string value)
#service_type = deckhand
[drydock]
@ -55,20 +53,10 @@
# From shipyard_airflow
#
# FQDN for the drydock service (string value)
#host = drydock-int.ucp
# Port for the drydock service (integer value)
#port = 9000
# TEMPORARY: password for drydock (string value)
#token = bigboss
# TEMPORARY: location of drydock yaml file (string value)
#site_yaml = /usr/local/airflow/plugins/drydock.yaml
# TEMPORARY: location of promenade yaml file (string value)
#prom_yaml = /usr/local/airflow/plugins/promenade.yaml
# The service type for the service playing the role of Drydock. The specified
# type is used to perform the service lookup in the Keystone service catalog.
# (string value)
#service_type = physicalprovisioner
[healthcheck]
@ -84,34 +72,6 @@
#endpoint = /api/v1.0/health
[keystone]
#
# From shipyard_airflow
#
# The url for OpenStack Authentication (string value)
#OS_AUTH_URL = http://keystone-api.ucp:80/v3
# OpenStack project name (string value)
#OS_PROJECT_NAME = service
# The OpenStack user domain name (string value)
#OS_USER_DOMAIN_NAME = Default
# The OpenStack username (string value)
#OS_USERNAME = shipyard
# THe OpenStack password for the shipyard svc acct (string value)
#OS_PASSWORD = password
# The OpenStack user domain name (string value)
#OS_REGION_NAME = Regionone
# The OpenStack identity api version (integer value)
#OS_IDENTITY_API_VERSION = 3
[keystone_authtoken]
#
@ -303,8 +263,7 @@
# From shipyard_airflow
#
# FQDN for the shipyard service (string value)
#host = shipyard-int.ucp
# Port for the shipyard service (integer value)
#port = 9000
# The service type for the service playing the role of Shipyard. The specified
# type is used to perform the service lookup in the Keystone service catalog.
# (string value)
#service_type = shipyard

134
shipyard_airflow/conf/config.py

@ -12,6 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
""" Module providing the oslo_config based configuration for Shipyard
"""
import logging
import keystoneauth1.loading as ks_loading
@ -75,14 +77,13 @@ SECTIONS = [
title='Shipyard connection info',
options=[
cfg.StrOpt(
'host',
default='shipyard-int.ucp',
help='FQDN for the shipyard service'
),
cfg.IntOpt(
'port',
default=9000,
help='Port for the shipyard service'
'service_type',
default='shipyard',
help=(
'The service type for the service playing the role '
'of Shipyard. The specified type is used to perform '
'the service lookup in the Keystone service catalog. '
)
),
]
),
@ -91,14 +92,13 @@ SECTIONS = [
title='Deckhand connection info',
options=[
cfg.StrOpt(
'host',
default='deckhand-int.ucp',
help='FQDN for the deckhand service'
),
cfg.IntOpt(
'port',
default=80,
help='Port for the deckhand service'
'service_type',
default='deckhand',
help=(
'The service type for the service playing the role '
'of Deckhand. The specified type is used to perform '
'the service lookup in the Keystone service catalog. '
)
),
]
),
@ -107,14 +107,13 @@ SECTIONS = [
title='Armada connection info',
options=[
cfg.StrOpt(
'host',
default='armada-int.ucp',
help='FQDN for the armada service'
),
cfg.IntOpt(
'port',
default=8000,
help='Port for the armada service'
'service_type',
default='armada',
help=(
'The service type for the service playing the role '
'of Armada. The specified type is used to perform '
'the service lookup in the Keystone service catalog. '
)
),
]
),
@ -123,32 +122,13 @@ SECTIONS = [
title='Drydock connection info',
options=[
cfg.StrOpt(
'host',
default='drydock-int.ucp',
help='FQDN for the drydock service'
),
cfg.IntOpt(
'port',
default=9000,
help='Port for the drydock service'
),
# TODO(Bryan Strassner) Remove this when integrated
cfg.StrOpt(
'token',
default='bigboss',
help='TEMPORARY: password for drydock'
),
# TODO(Bryan Strassner) Remove this when integrated
cfg.StrOpt(
'site_yaml',
default='/usr/local/airflow/plugins/drydock.yaml',
help='TEMPORARY: location of drydock yaml file'
),
# TODO(Bryan Strassner) Remove this when integrated
cfg.StrOpt(
'prom_yaml',
default='/usr/local/airflow/plugins/promenade.yaml',
help='TEMPORARY: location of promenade yaml file'
'service_type',
default='physicalprovisioner',
help=(
'The service type for the service playing the role '
'of Drydock. The specified type is used to perform '
'the service lookup in the Keystone service catalog. '
)
),
]
),
@ -168,56 +148,11 @@ SECTIONS = [
),
]
),
# TODO (Bryan Strassner) This section is in use by the operators we send
# to the airflow pod(s). Needs to be refactored out
# when those operators are updated.
ConfigSection(
name='keystone',
title='Keystone connection and credential information',
options=[
cfg.StrOpt(
'OS_AUTH_URL',
default='http://keystone-api.ucp:80/v3',
help='The url for OpenStack Authentication'
),
cfg.StrOpt(
'OS_PROJECT_NAME',
default='service',
help='OpenStack project name'
),
cfg.StrOpt(
'OS_USER_DOMAIN_NAME',
default='Default',
help='The OpenStack user domain name'
),
cfg.StrOpt(
'OS_USERNAME',
default='shipyard',
help='The OpenStack username'
),
cfg.StrOpt(
'OS_PASSWORD',
default='password',
help='THe OpenStack password for the shipyard svc acct'
),
cfg.StrOpt(
'OS_REGION_NAME',
default='Regionone',
help='The OpenStack user domain name'
),
cfg.IntOpt(
'OS_IDENTITY_API_VERSION',
default=3,
help='The OpenStack identity api version'
),
]
),
]
def register_opts(conf):
"""
Registers all the sections in this module.
""" Registers all the sections in this module.
"""
for section in SECTIONS:
conf.register_group(
@ -226,9 +161,6 @@ def register_opts(conf):
help=section.help))
conf.register_opts(section.options, group=section.name)
# TODO (Bryan Strassner) is there a better, more general way to do this,
# or is password enough? Probably need some guidance
# from someone with more experience in this space.
conf.register_opts(
ks_loading.get_auth_plugin_conf_options('password'),
group='keystone_authtoken'
@ -236,12 +168,16 @@ def register_opts(conf):
def list_opts():
""" List the options identified by this configuration
"""
return {
section.name: section.options for section in SECTIONS
}
def parse_args(args=None, usage=None, default_config_files=None):
""" Triggers the parsing of the arguments/configs
"""
CONF(args=args,
project='shipyard',
usage=usage,

28
shipyard_airflow/control/actions_api.py

@ -78,7 +78,6 @@ class ActionsResource(BaseResource):
# respond with the action and location for checking status
resp.status = falcon.HTTP_201
resp.body = self.to_json(action)
# TODO (Bryan Strassner) figure out the right way to do this:
resp.location = '/api/v1.0/actions/{}'.format(action['id'])
def create_action(self, action, context):
@ -245,16 +244,17 @@ class ActionsResource(BaseResource):
dag_id, self.to_json(conf_value)))
try:
resp = requests.get(req_url, timeout=15)
resp = requests.get(req_url, timeout=(5, 15))
self.info(context,
'Response code from Airflow trigger_dag: %s' %
resp.status_code)
# any 4xx/5xx will be HTTPError, which are RequestException
resp.raise_for_status()
response = resp.json()
self.info(context,
'Response from Airflow trigger_dag: %s' %
response)
except (RequestException) as rex:
except RequestException as rex:
self.error(context, "Request to airflow failed: %s" % rex.args)
raise ApiError(
title='Unable to complete request to Airflow',
@ -267,24 +267,10 @@ class ActionsResource(BaseResource):
}],
retry=True, )
# Returns error response if API call returns
# response code other than 200
if response["http_response_code"] != 200:
raise ApiError(
title='Unable to invoke workflow',
description=(
'Airflow URL not found by Shipyard.',
'Shipyard configuration is missing web_server value'),
status=falcon.HTTP_503,
error_list=[{
'message': response['output']
}],
retry=True, )
else:
dag_time = self._exhume_date(dag_id,
response['output']['stdout'])
dag_execution_date = dag_time.strftime('%Y-%m-%dT%H:%M:%S')
return dag_execution_date
dag_time = self._exhume_date(dag_id,
response['output']['stdout'])
dag_execution_date = dag_time.strftime('%Y-%m-%dT%H:%M:%S')
return dag_execution_date
def _exhume_date(self, dag_id, log_string):
# we are unable to use the response time because that

7
shipyard_airflow/control/api.py

@ -22,10 +22,14 @@ from shipyard_airflow.control.actions_steps_id_api import ActionsStepsResource
from shipyard_airflow.control.actions_validations_id_api import \
ActionsValidationsResource
from shipyard_airflow.control.base import BaseResource, ShipyardRequest
from shipyard_airflow.control.configdocs_api import (CommitConfigDocsResource,
ConfigDocsResource)
from shipyard_airflow.control.health import HealthResource
from shipyard_airflow.control.middleware import (AuthMiddleware,
ContextMiddleware,
LoggingMiddleware)
from shipyard_airflow.control.rendered_configdocs_api import \
RenderedConfigDocsResource
from shipyard_airflow.errors import (AppError, default_error_serializer,
default_exception_handler)
@ -55,6 +59,9 @@ def start_api():
ActionsStepsResource()),
('/actions/{action_id}/validations/{validation_id}',
ActionsValidationsResource()),
('/configdocs/{collection_id}', ConfigDocsResource()),
('/commitconfigdocs', CommitConfigDocsResource()),
('/renderedconfigdocs', RenderedConfigDocsResource()),
]
# Set up the 1.0 routes

217
shipyard_airflow/control/api_lock.py

@ -0,0 +1,217 @@
# Copyright 2017 AT&T Intellectual Property. All other rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
api_lock provides for a rudimentary lock mechanism using
the database to sync across multiple shipyard instances.
Also provided is the api_lock decorator to allow for resources
to easily declare the lock they should be able to acquire before
executing.
"""
from enum import Enum
import logging
from functools import wraps
import falcon
import ulid
from shipyard_airflow.db.db import SHIPYARD_DB
from shipyard_airflow.errors import ApiError
LOG = logging.getLogger(__name__)
def api_lock(api_lock_type):
"""
Decorator to handle allowing a resource method to institute a lock
based on the specified lock type.
These locks are intended for use around methods such as on_post
and on_get, etc...
"""
def lock_decorator(func):
@wraps(func)
def func_wrapper(self, req, resp, *args, **kwargs):
lock = ApiLock(api_lock_type,
req.context.external_marker,
req.context.user)
try:
lock.acquire()
return func(self, req, resp, *args, **kwargs)
except ApiLockAcquireError:
raise ApiError(
title='Blocked by another process',
description=(
'Another process is currently blocking this request '
'with a lock for {}. Lock expires in not more '
'than {} seconds'.format(
lock.lock_type_name,
lock.expires
)
),
status=falcon.HTTP_409,
retry=False,
)
finally:
lock.release()
return func_wrapper
return lock_decorator
class ApiLockType(Enum):
"""
ApiLockType defines the kinds of locks that can be set up using
this locking mechanism.
"""
CONFIGDOCS_UPDATE = {'name': 'configdocs_update', 'expires': 60}
class ApiLock(object):
"""
Api Lock provides for a simple locking mechanism for shipyard's
API classes. The lock provided is intended to block conflicting
activity across containers by using the backing database
to calculate if a lock is currently held.
The mechanism is as follows:
1) Attempt to write a lock record to the database such that:
there is no lock in the database of the same kind already that
is not either released or expired
1a) If the insert fails, the lock is not acquired.
2) Query the database for the latest lock of the type provided.
If the lock's id matches the ID of the process trying to
acquire the lock, the process should succeed.
If the lock ID doesn't match the record returned, the
process attempting to acquire a lock is blocked/failed.
(Note that the intended use is not to queue requests, but
rather fail them if something else holds the lock)
3) Upon completion of the activity, the lock holder will update
the lock record to indicate that it is released.
The database query used for insert will only insert if there
is not already an active lock record for the given type.
If the insert inserts zero rows, this indicates that the lock
is not acquired, and does not require a subsequent query.
The subsequent query is always used when the lock record insert
has been succesful, to handle race conditions. The select query
orders by both date and id, Whereby the randomness of the id
provides for a tiebreaker.
All locks expire based on their lock type, and default to
60 seconds
"""
def __init__(self,
api_lock_type,
reference_id,
user,
lock_db=SHIPYARD_DB):
"""
Set up the Api Lock, using the input ApiLockType.
Generates a ULID to represent this lock
:param api_lock_type: the ApiLockType for this lock
:param reference_id: the calling process' id provided for
purposes of correlation
:param user: the calling process' user for purposes of
tracking
"""
if (not isinstance(api_lock_type, ApiLockType) or
api_lock_type.value.get('name') is None):
raise ApiLockSetupError(
message='ApiLock requires a valid ApiLockType'
)
self.lock_id = ulid.ulid()
self.lock_type_name = api_lock_type.value.get('name')
self.expires = api_lock_type.value.get('expires', 60)
self.reference_id = reference_id
self.user = user
self.lock_db = lock_db
def acquire(self):
"""
Acquires a lock
Responds with an ApiLockAcquireError if the lock is not
acquired
"""
LOG.info('Acquiring lock type: %s. Lock id: %s.',
self.lock_type_name,
self.lock_id)
holds_lock = False
insert_worked = self.lock_db.insert_api_lock(
lock_id=self.lock_id,
lock_type=self.lock_type_name,
expires=self.expires,
user=self.user,
reference_id=self.reference_id
)
LOG.info('Insert lock %s %s',
self.lock_id,
'succeeded' if insert_worked else 'failed')
if insert_worked:
lock_retrieved = self.lock_db.get_api_lock(
lock_type=self.lock_type_name
)
holds_lock = lock_retrieved == self.lock_id
LOG.info(
'Lock %s is currently held. This lock is %s. Match=%s',
lock_retrieved,
self.lock_id,
holds_lock
)
if not holds_lock:
LOG.info('Api Lock not acquired')
raise ApiLockAcquireError()
def release(self):
"""
Release the lock
"""
try:
self.lock_db.release_api_lock(self.lock_id)
except Exception as error:
# catching Exception because this is a non-fatal case
# and has no expected action to be taken.
LOG.error('Exception raised during release of api lock: %s. '
'Unreleased lock for %s will expire in not more than '
'%s seconds. Exception: %s',
self.lock_id,
self.lock_type_name,
self.expires,
str(error))
class ApiLockError(Exception):
"""
Base exception for all api lock exceptions
"""
def __init__(self, message=None):
self.message = message
super().__init__()
class ApiLockSetupError(ApiLockError):
"""
Specifies that there was a problem during setup of the lock
"""
pass
class ApiLockAcquireError(ApiLockError):
"""
Signals to the calling process that this lock was not acquired
"""
pass

8
shipyard_airflow/control/base.py

@ -24,6 +24,10 @@ from shipyard_airflow.errors import InvalidFormatError
class BaseResource(object):
"""
The base resource for Shipyard entities/api handlers. This class
provides some reusable functionality.
"""
def __init__(self):
self.logger = logging.getLogger('shipyard.control')
@ -46,9 +50,7 @@ class BaseResource(object):
validation
"""
has_input = False
if ((req.content_length is not None or req.content_length != 0) and
(req.content_type is not None and
req.content_type.lower() == 'application/json')):
if req.content_length > 0 and 'application/json' in req.content_type:
raw_body = req.stream.read(req.content_length or 0)
if raw_body is not None:
has_input = True

178
shipyard_airflow/control/configdocs_api.py

@ -0,0 +1,178 @@
# Copyright 2017 AT&T Intellectual Property. All other rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Resources representing the configdocs API for shipyard
"""
import falcon
from oslo_config import cfg
from shipyard_airflow import policy
from shipyard_airflow.control import configdocs_helper
from shipyard_airflow.control.api_lock import (api_lock, ApiLockType)
from shipyard_airflow.control.base import BaseResource
from shipyard_airflow.control.configdocs_helper import (BufferMode,
ConfigdocsHelper)
from shipyard_airflow.errors import ApiError
CONF = cfg.CONF
VERSION_VALUES = ['buffer', 'committed']
class ConfigDocsResource(BaseResource):
"""
Configdocs handles the creation and retrieval of configuration
documents into Shipyard.
"""
@policy.ApiEnforcer('workflow_orchestrator:create_configdocs')
@api_lock(ApiLockType.CONFIGDOCS_UPDATE)
def on_post(self, req, resp, collection_id):
"""
Ingests a collection of documents
"""
document_data = req.stream.read(req.content_length or 0)
helper = ConfigdocsHelper(req.context.external_marker)
validations = self.post_collection(
helper=helper,
collection_id=collection_id,
document_data=document_data,
buffer_mode_param=req.params.get('buffermode')
)
resp.location = '/api/v1.0/configdocs/{}'.format(collection_id)
resp.body = self.to_json(validations)
resp.status = falcon.HTTP_201
@policy.ApiEnforcer('workflow_orchestrator:get_configdocs')
def on_get(self, req, resp, collection_id):
"""
Returns a collection of documents
"""
version = (req.params.get('version') or 'buffer')
self._validate_version_parameter(version)
helper = ConfigdocsHelper(req.context.external_marker)
# Not reformatting to JSON or YAML since just passing through
resp.body = self.get_collection(
helper=helper,
collection_id=collection_id,
version=version
)
resp.append_header('Content-Type', 'application/x-yaml')
resp.status = falcon.HTTP_200
def _validate_version_parameter(self, version):
# performs validation of version parameter
if version.lower() not in VERSION_VALUES:
raise ApiError(
title='Invalid version query parameter specified',
description=(
'version must be {}'.format(', '.join(VERSION_VALUES))
),
status=falcon.HTTP_400,
retry=False,
)
def get_collection(self,
helper,
collection_id,
version='buffer'):
"""
Attempts to retrieve the specified collection of documents
either from the buffer or committed version, as specified
"""
return helper.get_collection_docs(version, collection_id)
def post_collection(self,
helper,
collection_id,
document_data,
buffer_mode_param=None):
"""
Ingest the collection after checking preconditions
"""
if buffer_mode_param is None:
buffer_mode = BufferMode.REJECTONCONTENTS
else:
buffer_mode = ConfigdocsHelper.get_buffer_mode(buffer_mode_param)
if helper.is_buffer_valid_for_bucket(collection_id,
buffer_mode):
buffer_revision = helper.add_collection(
collection_id,
document_data
)
return helper.get_deckhand_validation_status(
buffer_revision
)
else:
raise ApiError(
title='Invalid collection specified for buffer',
description='Buffermode : {}'.format(buffer_mode.value),
status=falcon.HTTP_409,
error_list=[{
'message': ('Buffer is either not empty or the '
'collection already exists in buffer. '
'Setting a different buffermode may '
'provide the desired functionality')
}],
retry=False,
)
class CommitConfigDocsResource(BaseResource):
"""
Commits the buffered configdocs, if the validations pass (or are
overridden (force = true))
Returns the list of validations.
"""
unable_to_commmit = 'Unable to commit configuration documents'
@policy.ApiEnforcer('workflow_orchestrator:commit_configdocs')
@api_lock(ApiLockType.CONFIGDOCS_UPDATE)
def on_post(self, req, resp):
"""
Get validations from all UCP components
Functionality does not exist yet
"""
# force query parameter is False unless explicitly true
force = req.get_param_as_bool(name='force') or False
helper = ConfigdocsHelper(req.context.external_marker)
validations = self.commit_configdocs(helper, force)
resp.body = self.to_json(validations)
resp.status = validations.get('code', falcon.HTTP_200)
def commit_configdocs(self, helper, force):
"""
Attempts to commit the configdocs
"""
if helper.is_buffer_empty():
raise ApiError(
title=CommitConfigDocsResource.unable_to_commmit,
description='There are no documents in the buffer to commit',
status=falcon.HTTP_409,
retry=True
)
validations = helper.get_validations_for_buffer()
if force or validations.get('status') == 'Valid':
helper.tag_buffer(configdocs_helper.COMMITTED)
if force and validations.get('status') == 'Invalid':
# override the status in the response
validations['code'] = falcon.HTTP_200
if validations.get('message'):
validations['message'] = (
validations['message'] + ' FORCED SUCCESS'
)
else:
validations['message'] = 'FORCED SUCCESS'
return validations

609
shipyard_airflow/control/configdocs_helper.py

@ -0,0 +1,609 @@
# Copyright 2017 AT&T Intellectual Property. All other rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Configdocs helper primarily masquerades as a layer in front of
Deckhand, providing a representation of a buffer and a committed
bucket for Shipyard
"""
import enum
import json
import logging
import threading
import falcon
from oslo_config import cfg
import requests
from shipyard_airflow.control.deckhand_client import (
DeckhandClient,
DeckhandPaths,
DeckhandRejectedInputError,
DeckhandResponseError,
DocumentExistsElsewhereError,
NoRevisionsExistError
)
from shipyard_airflow.control.service_endpoints import (
Endpoints,
get_endpoint,
get_token
)
from shipyard_airflow.errors import ApiError, AppError
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
# keys for the revision dict, and consistency of strings for committed
# and buffer.
COMMITTED = 'committed'
BUFFER = 'buffer'
LATEST = 'latest'
REVISION_COUNT = 'count'
# string for rollback_commit consistency
ROLLBACK_COMMIT = 'rollback_commit'
class BufferMode(enum.Enum):
"""
Enumeration of the valid values for BufferMode
"""
REJECTONCONTENTS = 'rejectoncontents'
APPEND = 'append'
REPLACE = 'replace'
class ConfigdocsHelper(object):
"""
ConfigdocsHelper provides a layer to represent the buffer and committed
versions of design documents.
A new configdocs_helper is intended to be used for each invocation of the
service.
"""
def __init__(self, context_marker):
"""
Sets up this Configdocs helper with the supplied
context marker
"""
self.deckhand = DeckhandClient(context_marker)
self.context_marker = context_marker
# The revision_dict indicates the revisions that are
# associated with the buffered and committed doc sets. There
# is a risk of this being out of sync if there is high volume
# of commits and adds of docs going on in parallel, but not
# really much more than if this data is used in subsequent
# statements following a call within a method.
self.revision_dict = None
@staticmethod
def get_buffer_mode(buffermode_string):
"""
Checks the buffer mode for valid values.
"""
if buffermode_string:
try:
buffer_mode = BufferMode(buffermode_string.lower())
return buffer_mode
except ValueError:
return None
return BufferMode.REJECTONCONTENTS
def is_buffer_empty(self):
""" Check if the buffer is empty. """
return self._get_buffer_revision() is None
def is_collection_in_buffer(self, collection_id):
"""
Returns if the collection is represented in the buffer
"""
if self.is_buffer_empty():
return False
# If there is no committed revision, then it's 0.
# new revision is ok because we just checked for buffer emptiness
old_revision_id = self._get_committed_rev_id()
if old_revision_id is None:
old_revision_id = 0
try:
diff = self.deckhand.get_diff(
old_revision_id=old_revision_id,
new_revision_id=self._get_buffer_rev_id()
)
# the collection is in the buffer if it's not unmodified
return diff.get(collection_id, 'unmodified') != 'unmodified'
except DeckhandResponseError as drex:
raise AppError(
title='Unable to retrieve revisions',
description=(
'Deckhand has responded unexpectedly: {}:{}'.format(
drex.status_code,
drex.response_message
)
),
status=falcon.HTTP_500,
retry=False,
)
def is_buffer_valid_for_bucket(self, collection_id, buffermode):
"""
Indicates if the buffer as it currently is, may be written to
for the specified collection, based on the buffermode.
"""
# can always write if buffer is empty.
if self.is_buffer_empty():
return True
# from here down, the buffer is NOT empty.
# determine steps by the buffermode
if buffermode == BufferMode.REJECTONCONTENTS:
return False
elif buffermode == BufferMode.APPEND:
return not self.is_collection_in_buffer(collection_id)
# replace the buffer with last commit.
elif buffermode == BufferMode.REPLACE:
committed_rev_id = None
committed_rev = self._get_committed_revision()
if committed_rev:
committed_rev_id = committed_rev['id']
if committed_rev_id is None:
# TODO (bryan-strassner) use rollback to 0 if/when that
# is implemented in deckhand.
# reset_to_empty has side effect of deleting history
# from deckhand although it is only the corner case
# where no commit has ever been done.
self.deckhand.reset_to_empty()
else:
self.deckhand.rollback(committed_rev_id)
return True
def _get_revision_dict(self):
"""
Returns a dictionary with values representing the revisions in
Deckhand that Shipyard cares about - committed, buffer,
and latest, as well as a count of revisions
Committed and buffer are revisions associated with the
shipyard tags. If either of those are not present in deckhand,
returns None for the value.
Latest holds the revision information for the newest revision.
"""
# return the cached instance version of the revision dict.
if self.revision_dict is not None:
return self.revision_dict
# or generate one for the cache
committed_revision = None
buffer_revision = None
latest_revision = None
revision_count = 0
try:
revisions = self.deckhand.get_revision_list()
revision_count = len(revisions)
if revisions:
latest_revision = revisions[-1]
for revision in reversed(revisions):
tags = revision.get('tags', [])
if COMMITTED in tags or ROLLBACK_COMMIT in tags:
committed_revision = revision
break
else:
# there are buffer revisions, only grab it on
# the first pass through
# if the first revision is committed, or if there
# are no revsisions, buffer revsision stays None
if buffer_revision is None:
buffer_revision = revision
except NoRevisionsExistError:
# the values of None/None/None/0 are fine
pass
except DeckhandResponseError as drex:
raise AppError(
title='Unable to retrieve revisions',
description=(
'Deckhand has responded unexpectedly: {}:{}'.format(
drex.status_code,
drex.response_message
)
),
status=falcon.HTTP_500,
retry=False,
)
self.revision_dict = {
COMMITTED: committed_revision,
BUFFER: buffer_revision,
LATEST: latest_revision,
REVISION_COUNT: revision_count
}
return self.revision_dict
def _get_buffer_revision(self):
# convenience helper to drill down to Buffer revision
return self._get_revision_dict().get(BUFFER)
def _get_buffer_rev_id(self):
# convenience helper to drill down to Buffer revision id
buf_rev = self._get_revision_dict().get(BUFFER)
return buf_rev['id'] if buf_rev else None
def _get_latest_revision(self):
# convenience helper to drill down to latest revision
return self._get_revision_dict().get(LATEST)
def _get_latest_rev_id(self):
# convenience helper to drill down to latest revision id
latest_rev = self._get_revision_dict().get(LATEST)
return latest_rev['id'] if latest_rev else None
def _get_committed_revision(self):
# convenience helper to drill down to committed revision
return self._get_revision_dict().get(COMMITTED)
def _get_committed_rev_id(self):
# convenience helper to drill down to committed revision id
committed_rev = self._get_revision_dict().get(COMMITTED)
return committed_rev['id'] if committed_rev else None
def get_collection_docs(self, version, collection_id):
"""
Returns the requested collection of docs based on the version
specifier. Since the default is the buffer, only return
committed if explicitly stated. No need to further check the
parameter for validity here.
"""
LOG.info('Retrieving collection %s from %s', collection_id, version)
if version == COMMITTED:
return self._get_committed_docs(collection_id)
return self._get_doc_from_buffer(collection_id)
def _get_doc_from_buffer(self, collection_id):
"""
Returns the collection if it exists in the buffer.
If the buffer contains the collection, the latest
representation is what we want.
"""
# Need to guard with this check for buffer to ensure
# that the collection is not just carried through unmodified
# into the buffer, and is actually represented.
if self.is_collection_in_buffer(collection_id):
# prior check for collection in buffer means the buffer
# revision exists
buffer_id = self._get_buffer_rev_id()
return self.deckhand.get_docs_from_revision(
revision_id=buffer_id,
bucket_id=collection_id
)
raise ApiError(
title='No documents to retrieve',
description=('The Shipyard buffer is empty or does not contain '
'this collection'),
status=falcon.HTTP_404,
retry=False,
)
def _get_committed_docs(self, collection_id):
"""
Returns the collection if it exists as committed.
"""
committed_id = self._get_committed_rev_id()
if committed_id:
return self.deckhand.get_docs_from_revision(
revision_id=committed_id,
bucket_id=collection_id
)
# if there is no committed...
raise ApiError(
title='No documents to retrieve',
description='There is no committed version of this collection',
status=falcon.HTTP_404,
retry=False,
)
def get_rendered_configdocs(self, version=BUFFER):
"""
Returns the rendered configuration documents for the specified
revision (by name BUFFER, COMMITTED)
"""
revision_dict = self._get_revision_dict()
if version in (BUFFER, COMMITTED):
if revision_dict.get(version):
revision_id = revision_dict.get(version).get('id')
return self.deckhand.get_rendered_docs_from_revision(
revision_id=revision_id
)
else:
raise ApiError(
title='This revision does not exist',
description='{} version does not exist'.format(version),
status=falcon.HTTP_404,
retry=False,
)
def get_validations_for_buffer(self):
"""
Convenience method to do validations for buffer version.
"""
buffer_rev_id = self._get_buffer_rev_id()
if buffer_rev_id:
return self.get_validations_for_revision(buffer_rev_id)
raise AppError(
title='Unable to start validation of buffer',
description=('Buffer revision id could not be determined from'
'Deckhand'),
status=falcon.HTTP_500,
retry=False,
)
@staticmethod
def _get_design_reference(revision_id):
# Constructs the design reference as json for use by other components
design_reference = {
"rel": "design",
"href": "deckhand+{}/rendered-documents".format(
DeckhandPaths.RENDERED_REVISION_DOCS.value.format(revision_id)
),
"type": "application/x-yaml"
}
return json.dumps(design_reference)
@staticmethod
def _get_validation_endpoints():
# returns the list of validation endpoint supported
val_ep = '{}/validatedesign'
return [
{'name': 'Drydock',
'url': val_ep.format(get_endpoint(Endpoints.DRYDOCK))},
{'name': 'Armada',
'url': val_ep.format(get_endpoint(Endpoints.ARMADA))},
]
@staticmethod
def _get_validation_threads(validation_endpoints,
revision_id,
context_marker):
# create a list of validation threads from the endpoints
validation_threads = []
for endpoint in validation_endpoints:
# create a holder for things we need back from the threads
response = {'response': None}
exception = {'exception': None}
validation_threads.append(
{
'thread': threading.Thread(
target=ConfigdocsHelper._get_validations_for_component,
args=(
endpoint['url'],
ConfigdocsHelper._get_design_reference(
revision_id
),
response,
exception,
context_marker
)
),
'name': endpoint['name'],
'url': endpoint['url'],
'response': response,
'exception': exception
}
)
return validation_threads
@staticmethod
def _get_validations_for_component(url,
design_reference,
response,
exception,
context_marker):
# Invoke the POST for validation
try:
headers = {
'X-Context-Marker': context_marker,
'X-Auth-Token': get_token(),
'content-type': 'application/x-yaml'
}