Merge "Delete old policies when upgrading to HTTPS"

This commit is contained in:
Jenkins 2016-09-06 18:06:50 +00:00 committed by Gerrit Code Review
commit 12188071bc
14 changed files with 407 additions and 17 deletions

View File

@ -0,0 +1,66 @@
# Copyright (c) 2016 Rackspace, Inc.
#
# 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.
import time
from oslo_config import cfg
from oslo_log import log
from poppy import bootstrap
from poppy.common import cli
LOG = log.getLogger(__name__)
CLI_OPT = [
cfg.BoolOpt(
'run_as_daemon',
default=True,
required=False,
help='Run this script as a long running process.'
),
cfg.IntOpt(
'sleep_interval',
default=60,
required=False,
help='Sleep interval between runs of http policy delete.'
),
]
@cli.runnable
def run():
# TODO(kgriffs): For now, we have to use the global config
# to pick up common options from openstack.common.log, since
# that module uses the global CONF instance exclusively.
conf = cfg.ConfigOpts()
conf.register_cli_opts(CLI_OPT)
log.register_options(conf)
conf(project='poppy', prog='poppy')
log.setup(conf, 'poppy')
server = bootstrap.Bootstrap(conf)
sleep_interval = conf.sleep_interval
while True:
(
run_list,
ignore_list
) = server.manager.background_job_controller.delete_http_policy()
LOG.info(
"Policies, attempting to delete {0}, ignored {0}".format(
run_list, ignore_list))
if conf.run_as_daemon is False:
break
time.sleep(sleep_interval)

View File

@ -14,6 +14,7 @@
# limitations under the License.
import cgi
import itertools
import pprint
from oslo_log import log
@ -54,3 +55,14 @@ def help_escape(potentially_bad_string):
if potentially_bad_string is None:
LOG.warning('Should not happen: trying to escape a None object')
return cgi.escape(potentially_bad_string or "")
# remove duplicates
# see http://bit.ly/1mX2Vcb for details
def remove_duplicates(data):
"""Remove duplicates from the data (normally a list).
The data must be sortable and have an equality operator
"""
data = sorted(data)
return [k for k, _ in itertools.groupby(data)]

View File

@ -17,11 +17,14 @@ import json
from oslo_log import log
from poppy.common import util
from poppy.manager import base
from poppy.model import ssl_certificate
from poppy.notification.mailgun import driver as n_driver
from poppy.provider.akamai.background_jobs.check_cert_status_and_update \
import check_cert_status_and_update_flow
from poppy.provider.akamai.background_jobs.delete_policy \
import delete_obsolete_http_policy_flow
from poppy.provider.akamai.background_jobs.update_property import \
update_property_flow
from poppy.provider.akamai import driver as a_driver
@ -286,3 +289,59 @@ class BackgroundJobController(base.BackgroundJobController):
deleted = tuple(x for x in orig if x not in res)
# other provider's san_mapping_list implementation goes here
return res, deleted
def delete_http_policy(self):
http_policies = []
run_list = []
ignore_list = []
if 'akamai' in self._driver.providers:
akamai_driver = self._driver.providers['akamai'].obj
http_policies += akamai_driver.http_policy_queue.traverse_queue(
consume=True
)
http_policies = [json.loads(x) for x in http_policies]
http_policies = util.remove_duplicates(http_policies)
for policy_dict in http_policies:
cert_for_domain = self.cert_storage.get_certs_by_domain(
policy_dict['policy_name'],
project_id=policy_dict['project_id'],
cert_type='san'
)
if cert_for_domain == []:
ignore_list.append(policy_dict)
LOG.info(
"No cert found for policy name. "
"Policy {0} won't persist on the queue. ".format(
policy_dict
)
)
continue
if cert_for_domain.get_cert_status() != 'deployed':
ignore_list.append(policy_dict)
LOG.info(
"Policy {0} is not ready for deletion. "
"Certificate exists but hasn't deployed yet. "
"Sending back to queue for later retry.".format(
policy_dict
)
)
akamai_driver.http_policy_queue.enqueue_http_policy(
json.dumps(policy_dict)
)
continue
kwargs = {
'configuration_number': policy_dict[
'configuration_number'],
'policy_name': policy_dict['policy_name']
}
self.distributed_task_controller.submit_task(
delete_obsolete_http_policy_flow.
delete_obsolete_http_policy,
**kwargs
)
run_list.append(policy_dict)
return run_list, ignore_list

View File

@ -13,13 +13,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import itertools
import json
from oslo_context import context as context_utils
from oslo_log import log
from poppy.common import errors
from poppy.common import util
from poppy.distributed_task.taskflow.flow import create_ssl_certificate
from poppy.distributed_task.taskflow.flow import delete_ssl_certificate
from poppy.distributed_task.taskflow.flow import recreate_ssl_certificate
@ -176,16 +176,7 @@ class DefaultSSLCertificateController(base.SSLCertificateController):
res = akamai_driver.mod_san_queue.dequeue_mod_san_request()
retry_list.append(json.loads(res.decode('utf-8')))
# remove duplicates
# see http://bit.ly/1mX2Vcb for details
def remove_duplicates(data):
"""Remove duplicates from the data (normally a list).
The data must be sortable and have an equality operator
"""
data = sorted(data)
return [k for k, _ in itertools.groupby(data)]
retry_list = remove_duplicates(retry_list)
retry_list = util.remove_duplicates(retry_list)
# double check in POST. This check should really be first done in
# PUT

View File

@ -0,0 +1,34 @@
# Copyright (c) 2016 Rackspace, Inc.
#
# 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.
from oslo_config import cfg
from oslo_log import log
from taskflow.patterns import linear_flow
from poppy.provider.akamai.background_jobs.delete_policy import (
delete_obsolete_http_policy_tasks)
LOG = log.getLogger(__name__)
conf = cfg.CONF
conf(project='poppy', prog='poppy', args=[])
def delete_obsolete_http_policy():
flow = linear_flow.Flow('Deleting obsolete HTTP policy').add(
delete_obsolete_http_policy_tasks.DeleteObsoleteHTTPPolicy(),
)
return flow

View File

@ -0,0 +1,56 @@
# Copyright (c) 2016 Rackspace, Inc.
#
# 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.
from oslo_config import cfg
from oslo_log import log
from taskflow import task
from poppy.distributed_task.utils import memoized_controllers
LOG = log.getLogger(__name__)
conf = cfg.CONF
conf(project='poppy', prog='poppy', args=[])
class DeleteObsoleteHTTPPolicy(task.Task):
"""Delete old HTTP policy once a domain is upgraded to HTTPS SAN."""
def __init__(self):
super(DeleteObsoleteHTTPPolicy, self).__init__()
service_controller, self.providers = \
memoized_controllers.task_controllers('poppy', 'providers')
self.akamai_driver = self.providers['akamai'].obj
def execute(self, configuration_number, policy_name):
"""Deletes old HTTP policy once a domain is upgraded to HTTPS+san.
:param configuration_number: akamai configuration number
:param policy_name: name of policy on akamai policy api
"""
resp = self.akamai_driver.policy_api_client.delete(
self.akamai_driver.akamai_policy_api_base_url.format(
configuration_number=configuration_number,
policy_name=policy_name
)
)
LOG.info(
'akamai response code: {0}'.format(resp.status_code))
LOG.info('akamai response text: {0}'.format(resp.text))
if resp.status_code != 200:
raise RuntimeError(resp.text)
LOG.info(
'Delete old policy {0} complete'.format(policy_name))

View File

@ -27,6 +27,7 @@ from poppy.common import decorators
from poppy.provider.akamai import controllers
from poppy.provider.akamai.domain_san_mapping_queue import zk_san_mapping_queue
from poppy.provider.akamai import geo_zone_code_mapping
from poppy.provider.akamai.http_policy_queue import http_policy_queue
from poppy.provider.akamai.mod_san_queue import zookeeper_queue
from poppy.provider import base
import uuid
@ -200,6 +201,9 @@ class CDNProvider(base.Driver):
self.san_mapping_queue = zk_san_mapping_queue.ZookeeperSanMappingQueue(
self._conf
)
self.http_policy_queue = http_policy_queue.ZookeeperHttpPolicyQueue(
self._conf
)
self.metrics_resolution = self.akamai_conf.metrics_resolution

View File

@ -0,0 +1,58 @@
# Copyright (c) 2016 Rackspace, Inc.
#
# 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.
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class HttpPolicyQueue(object):
"""Keep track of old HTTP policies for deletion.
The policy object on queue is used to kick off a task
that deletes obsolete akamai http policies at pre-defined intervals.
"""
def __init__(self, conf):
self._conf = conf
def enqueue_http_policy(self, http_policy):
"""Add new http policy element to the queue.
:param http_policy: a new element to add to the queue
:type http_policy: dict
"""
raise NotImplementedError
def dequeue_http_policy(self, consume=True):
"""Remove and return an item from the queue.
:param consume: if true the policy is removed from the list and
returned otherwise the policy is retrieved queue
"""
raise NotImplementedError
def traverse_queue(self, consume=False):
"""Traverse queue and return all items on the queue in a list"""
raise NotImplementedError
def put_queue_data(self, queue_data_list):
"""Clear the queue and put new queue data list in the queue.
:param queue_data_list: new queue data to replace current queue data
:type queue_data_list: [dict()] -- list of dictionaries
"""
raise NotImplementedError

View File

@ -0,0 +1,96 @@
# Copyright (c) 2016 Rackspace, Inc.
#
# 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.
from kazoo.recipe import queue
from oslo_config import cfg
from poppy.common import decorators
from poppy.provider.akamai.http_policy_queue import base
from poppy.provider.akamai import utils
AKAMAI_OPTIONS = [
# queue backend configs
cfg.StrOpt(
'queue_backend_type',
help='SAN Cert Queueing backend'),
cfg.ListOpt(
'queue_backend_host',
default=['localhost'],
help='default queue backend server hosts'),
cfg.IntOpt(
'queue_backend_port',
default=2181,
help='default'
' default queue backend server port (e.g: 2181)'),
cfg.StrOpt(
'http_policy_queue_path',
default='/http_policy_queue',
help='Zookeeper path '
'for http_policy_queue'
),
]
AKAMAI_GROUP = 'drivers:provider:akamai:queue'
class ZookeeperHttpPolicyQueue(base.HttpPolicyQueue):
def __init__(self, conf):
super(ZookeeperHttpPolicyQueue, self).__init__(conf)
self._conf.register_opts(AKAMAI_OPTIONS,
group=AKAMAI_GROUP)
self.akamai_conf = self._conf[AKAMAI_GROUP]
@decorators.lazy_property(write=False)
def http_policy_queue_backend(self):
return queue.LockingQueue(
self.zk_client,
self.akamai_conf.http_policy_queue_path)
@decorators.lazy_property(write=False)
def zk_client(self):
return utils.connect_to_zookeeper_queue_backend(self.akamai_conf)
def enqueue_http_policy(self, http_policy):
self.http_policy_queue_backend.put(http_policy)
def traverse_queue(self, consume=False):
res = []
while len(self.http_policy_queue_backend) > 0:
item = self.http_policy_queue_backend.get()
self.http_policy_queue_backend.consume()
res.append(item)
if consume is False:
self.http_policy_queue_backend.put_all(res)
return res
def put_queue_data(self, queue_data):
# put queue data will replace all existing
# queue data with the incoming new queue_data
# dequeue all the existing data
while len(self.http_policy_queue_backend) > 0:
self.http_policy_queue_backend.get()
self.http_policy_queue_backend.consume()
# put in all the new data
self.http_policy_queue_backend.put_all(queue_data)
return queue_data
def dequeue_http_policy(self, consume=True):
res = self.http_policy_queue_backend.get()
if consume:
self.http_policy_queue_backend.consume()
return res

View File

@ -33,6 +33,10 @@ LOG = log.getLogger(__name__)
class ServiceController(base.ServiceBase):
@property
def http_policy_queue(self):
return self.driver.http_policy_queue
@property
def policy_api_client(self):
return self.driver.policy_api_client
@ -459,18 +463,26 @@ class ServiceController(base.ServiceBase):
):
is_upgrade = True
# skip policy delete if a http -> https+san
# upgrade is detected
configuration_number = self._get_configuration_number(
util.dict2obj(policy))
# when an upgrade is detected, keep track of the
# old http policy. the old http policy will be deleted
# later.
if is_upgrade is True:
LOG.info(
"{0} was upgraded from http to https san. "
"Skipping old policy delete.".format(
"Queuing old http policy for delete.".format(
policy['policy_name']))
self.http_policy_queue.enqueue_http_policy(
json.dumps({
'configuration_number': configuration_number,
'policy_name': policy['policy_name'],
'project_id': service_obj.project_id
})
)
continue
configuration_number = self._get_configuration_number(
util.dict2obj(policy))
LOG.info('Starting to delete old policy %s' %
policy['policy_name'])
resp = self.policy_api_client.delete(

View File

@ -35,6 +35,7 @@ console_scripts =
poppy-worker = poppy.cmd.task_flow_worker:run
akamai-cert-status-check-and-update = poppy.cmd.akamai_check_and_update_cert_status:run
akamai-property-udpate-mod-san = poppy.cmd.akamai_update_papi_property_for_mod_san:run
delete-obsolete-http-policy = poppy.cmd.delete_obsolete_http_policies:run
poppy.transport =
pecan = poppy.transport.pecan:Driver

View File

@ -49,6 +49,7 @@ class TestServices(base.TestCase):
self.ccu_client = self.driver.ccu_api_client
self.driver.provider_name = 'Akamai'
self.driver.http_conf_number = 1
self.driver.akamai_https_access_url_suffix = str(uuid.uuid1())
self.san_cert_cnames = [str(x) for x in range(7)]
self.driver.san_cert_cnames = self.san_cert_cnames