Add cassandra akamai san info storage

Change-Id: I1a37130190399e6c5f33148c258b4b7fa3ddeaf8

etc/poppy.conf

@@ -8,6 +8,9 @@
 # Datacenter in which the API is hosted.
 datacenter = DC1
 
+# whether to use the same storage config from [drivers:storage:<storage_name>]
+use_same_storage_driver = True
+
 # Show debugging output in logs (sets DEBUG log level output)
 ;debug = False
 
@@ -163,12 +166,39 @@ san_cert_hostname_limit = "MY_SAN_HOSTNAMES_LMIT"
 contract_id = "MY_CONTRACT_ID"
 group_id = "MY_GROUP_ID"
 property_id = "MY_PROPERTY_ID"
+# akamai_san_info_storage driver module (e.g. zookeeper, cassandra)
+san_info_storage_type = cassandra
 
 [drivers:provider:akamai:storage]
-storage_backend_type = zookeeper
-storage_backend_host = <your_transport_server(s)>
-storage_backend_port = <your_transport_port>
-san_info_storage_path = '/san_info'
+# Zookeeper san_info_storage_type config options
+#storage_backend_type = zookeeper
+#storage_backend_host = <your_transport_server(s)>
+#storage_backend_port = <your_transport_port>
+#san_info_storage_path = '/san_info'
+
+# Comma-separated list of hosts (Example: cass01,cass02,cass03)
+cluster = localhost
+;port = 9042
+ssl_enabled = False
+ssl_ca_certs = </absolute/path/to/cassandra.crt>
+auth_enabled = False
+username = cassandra_username
+password = cassandra_password
+# Either RoundRobinPolicy or DCAwareRoundRobinPolicy. DCAwareRoundRobinPolicy
+# requires the datacenter option in [DEFAULT] to be configured.
+load_balance_strategy = RoundRobinPolicy
+consistency_level = ONE
+migrations_consistency_level = LOCAL_QUORUM
+max_schema_agreement_wait = 30
+keyspace = akamai_san_info
+# Replication strategy to use for the keyspace. This value is plugged into
+# `map` as show in the syntax here: http://www.datastax.com/documentation/cql/3
+# .1/cql/cql_reference/create_keyspace_r.html
+replication_strategy = class:SimpleStrategy, replication_factor:1
+# Path to directory containing CQL migration scripts
+migrations_path = <poppy_code_path>/poppy/storage/cassandra/migrations
+
+
 
 [drivers:provider:akamai:queue]
 queue_backend_type = zookeeper

poppy/provider/akamai/driver.py

@@ -20,11 +20,12 @@ import json
 from akamai import edgegrid
 from oslo_config import cfg
 import requests
+from stevedore import driver
 
+from poppy.common import decorators
 from poppy.openstack.common import log
 from poppy.provider.akamai import controllers
 from poppy.provider.akamai.mod_san_queue import zookeeper_queue
-from poppy.provider.akamai.san_info_storage import zookeeper_storage
 from poppy.provider import base
 
 LOG = log.getLogger(__name__)
@@ -91,6 +92,8 @@ AKAMAI_OPTIONS = [
     cfg.IntOpt('san_cert_hostname_limit', default=80,
                help='default limit on how many hostnames can'
                ' be held by a SAN cert'),
+    cfg.StrOpt('san_info_storage_type',
+               help='Storage type for storing san cert information'),
 
     # related info for SPS && PAPI APIs
     cfg.StrOpt(
@@ -161,16 +164,29 @@ class CDNProvider(base.Driver):
             )
         ])
 
+        self.akamai_sps_api_client = self.akamai_policy_api_client
+
         self.san_cert_cnames = self.akamai_conf.san_cert_cnames
         self.san_cert_hostname_limit = self.akamai_conf.san_cert_hostname_limit
 
-        self.akamai_sps_api_client = self.akamai_policy_api_client
-
-        self.san_info_storage = (
-            zookeeper_storage.ZookeeperSanInfoStorage(self._conf))
         self.mod_san_queue = (
             zookeeper_queue.ZookeeperModSanQueue(self._conf))
 
+    @decorators.lazy_property(write=False)
+    def san_info_storage(self):
+        storage_backend_type = 'poppy.provider.akamai.san_info_storage'
+        storage_backend_name = self.akamai_conf.san_info_storage_type
+
+        args = [self._conf]
+
+        san_info_storage = driver.DriverManager(
+            namespace=storage_backend_type,
+            name=storage_backend_name,
+            invoke_on_load=True,
+            invoke_args=args)
+
+        return san_info_storage.driver
+
     def is_alive(self):
 
         request_headers = {

poppy/provider/akamai/san_info_storage/base.py

@@ -27,11 +27,18 @@ class BaseAkamaiSanInfoStorage(object):
     def __init__(self, conf):
         self._conf = conf
 
+    @abc.abstractmethod
     def get_cert_info(self, san_cert_name):
         raise NotImplementedError
 
+    @abc.abstractmethod
     def save_cert_last_spsid(self, san_cert_name, sps_id_value):
         raise NotImplementedError
 
+    @abc.abstractmethod
     def get_cert_last_spsid(self, san_cert_name):
         raise NotImplementedError
+
+    @abc.abstractmethod
+    def list_all_san_cert_names(self):
+        raise NotImplementedError

poppy/provider/akamai/san_info_storage/cassandra_storage.py

@@ -0,0 +1,316 @@
+# Copyright (c) 2015 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 copy
+import json
+import os
+import ssl
+
+import cassandra
+from cassandra import auth
+from cassandra import cluster
+from cassandra import policies
+from cassandra import query
+from cdeploy import migrator
+from oslo_config import cfg
+
+from poppy.common import decorators
+from poppy.openstack.common import log as logging
+from poppy.provider.akamai.san_info_storage import base
+
+
+_DEFAULT_OPTIONS = [
+    cfg.StrOpt('datacenter', default='',
+               help='Host datacenter of the API'),
+    cfg.BoolOpt('use_same_storage_driver', default=True,
+                help='Whether to use the same poppy storage driver')
+]
+
+CASSANDRA_OPTIONS = [
+    cfg.ListOpt('cluster', default=['127.0.0.1'],
+                help='Cassandra cluster contact points'),
+    cfg.IntOpt('port', default=9042, help='Cassandra cluster port'),
+    cfg.BoolOpt('ssl_enabled', default=False,
+                help='Communicate with Cassandra over SSL?'),
+    cfg.StrOpt('ssl_ca_certs', default='',
+               help='Absolute path to the appropriate .crt file'),
+    cfg.BoolOpt('auth_enabled', default=False,
+                help='Does Cassandra have authentication enabled?'),
+    cfg.StrOpt('username', default='', help='Cassandra username'),
+    cfg.StrOpt('password', default='', help='Cassandra password'),
+    cfg.StrOpt('load_balance_strategy', default='RoundRobinPolicy',
+               help='Load balancing strategy for connecting to cluster nodes'),
+    cfg.StrOpt('consistency_level', default='ONE',
+               help='Consistency level of your cassandra query'),
+    cfg.StrOpt('migrations_consistency_level', default='LOCAL_QUORUM',
+               help='Consistency level of cassandra migration queries'),
+    cfg.IntOpt('max_schema_agreement_wait', default=10,
+               help='The maximum duration (in seconds) that the driver will'
+               ' wait for schema agreement across the cluster.'),
+    cfg.StrOpt('keyspace', default='poppy',
+               help='Keyspace for all queries made in session'),
+    cfg.DictOpt(
+        'replication_strategy',
+        default={
+            'class': 'SimpleStrategy',
+            'replication_factor': '1'
+        },
+        help='Replication strategy for Cassandra cluster'
+    ),
+    cfg.StrOpt(
+        'migrations_path',
+        default=os.path.join(os.path.dirname(
+            os.path.dirname(
+                os.path.dirname(
+                    os.path.dirname(__file__)))),
+            'storage',
+            'cassandra',
+            'migrations'),
+        help='Path to directory containing CQL migration scripts',
+    )
+]
+
+AKAMAI_CASSANDRA_STORAGE_GROUP = 'drivers:provider:akamai:storage'
+
+LOG = logging.getLogger(__name__)
+
+
+GET_PROVIDER_INFO = '''
+    SELECT info from providers_info
+    WHERE provider_name = %(provider_name)s
+'''
+
+UPDATE_PROVIDER_INFO = '''
+    UPDATE providers_info
+    set info = %(info)s
+    WHERE provider_name = %(provider_name)s
+'''
+
+CREATE_PROVIDER_INFO = '''
+    INSERT INTO providers_info (
+        provider_name,
+        info
+        )
+    VALUES (%(provider_name)s,
+        %(info)s)
+'''
+
+
+def _connection(conf, datacenter, keyspace=None):
+    """connection.
+
+    :param datacenter
+    :returns session
+    """
+    ssl_options = None
+    if conf.ssl_enabled:
+        ssl_options = {
+            'ca_certs': conf.ssl_ca_certs,
+            'ssl_version': ssl.PROTOCOL_TLSv1
+        }
+
+    auth_provider = None
+    if conf.auth_enabled:
+        auth_provider = auth.PlainTextAuthProvider(
+            username=conf.username,
+            password=conf.password
+        )
+
+    load_balancing_policy_class = getattr(policies, conf.load_balance_strategy)
+    if load_balancing_policy_class is policies.DCAwareRoundRobinPolicy:
+        load_balancing_policy = load_balancing_policy_class(datacenter)
+    else:
+        load_balancing_policy = load_balancing_policy_class()
+
+    cluster_connection = cluster.Cluster(
+        conf.cluster,
+        auth_provider=auth_provider,
+        load_balancing_policy=load_balancing_policy,
+        port=conf.port,
+        ssl_options=ssl_options,
+        max_schema_agreement_wait=conf.max_schema_agreement_wait
+    )
+
+    session = cluster_connection.connect()
+    if not keyspace:
+        keyspace = conf.keyspace
+    try:
+        session.set_keyspace(keyspace)
+    except cassandra.InvalidRequest:
+        _create_keyspace(session, keyspace, conf.replication_strategy)
+
+    migration_session = copy.copy(session)
+    migration_session.default_consistency_level = \
+        getattr(cassandra.ConsistencyLevel, conf.migrations_consistency_level)
+    _run_migrations(keyspace, conf.migrations_path, migration_session)
+
+    session.row_factory = query.dict_factory
+
+    return session
+
+
+def _create_keyspace(session, keyspace, replication_strategy):
+    """create_keyspace.
+
+    :param keyspace
+    :param replication_strategy
+    """
+    LOG.debug('Creating keyspace: ' + keyspace)
+
+    # replication factor will come in as a string with quotes already
+    session.execute(
+        "CREATE KEYSPACE " + keyspace + " " +
+        "WITH REPLICATION = " + str(replication_strategy) + ";"
+    )
+    session.set_keyspace(keyspace)
+
+
+def _run_migrations(keyspace, migrations_path, session):
+    LOG.debug('Running schema migration(s) on keyspace: %s' % keyspace)
+
+    schema_migrator = migrator.Migrator(migrations_path, session)
+    schema_migrator.run_migrations()
+
+
+class CassandraSanInfoStorage(base.BaseAkamaiSanInfoStorage):
+
+    def __init__(self, conf):
+        super(CassandraSanInfoStorage, self).__init__(conf)
+
+        self._conf.register_opts(_DEFAULT_OPTIONS)
+        if self._conf.use_same_storage_driver:
+            from poppy.storage.cassandra import driver
+            self._conf.register_opts(driver.CASSANDRA_OPTIONS,
+                                     group=AKAMAI_CASSANDRA_STORAGE_GROUP)
+        else:
+            self._conf.register_opts(CASSANDRA_OPTIONS,
+                                     group=AKAMAI_CASSANDRA_STORAGE_GROUP)
+        self.cassandra_conf = self._conf[AKAMAI_CASSANDRA_STORAGE_GROUP]
+        self.datacenter = conf.datacenter
+        self.consistency_level = getattr(
+            cassandra.ConsistencyLevel,
+            conf[AKAMAI_CASSANDRA_STORAGE_GROUP].consistency_level)
+
+    @decorators.lazy_property(write=False)
+    def connection(self):
+        return _connection(self.cassandra_conf, self.datacenter)
+
+    @property
+    def session(self):
+        return self.connection
+
+    def _get_akamai_provider_info(self):
+        args = {
+            "provider_name": 'akamai'
+        }
+
+        stmt = query.SimpleStatement(
+            GET_PROVIDER_INFO,
+            consistency_level=self.consistency_level)
+        results = self.session.execute(stmt, args)
+
+        if len(results) != 1:
+            raise ValueError('No akamai providers info found.')
+
+        result = results[0]
+
+        return result
+
+    def _get_akamai_san_certs_info(self):
+        return json.loads(self._get_akamai_provider_info()['info']['san_info'])
+
+    def list_all_san_cert_names(self):
+        return self._get_akamai_san_certs_info().keys()
+
+    def get_cert_info(self, san_cert_name):
+        the_san_cert_info = self._get_akamai_san_certs_info().get(
+            san_cert_name
+        )
+
+        if the_san_cert_info is None:
+            raise ValueError('No san cert info found for %s.' % san_cert_name)
+
+        jobId = the_san_cert_info.get("jobId")
+        issuer = the_san_cert_info.get("issuer")
+        ipVersion = the_san_cert_info.get("ipVersion")
+        slot_deployment_klass = the_san_cert_info.get("slot_deployment_klass")
+
+        res = {
+            # This will always be the san cert name
+            'cnameHostname': san_cert_name,
+            'jobId': jobId,
+            'issuer': issuer,
+            'createType': 'modSan',
+            'ipVersion': ipVersion,
+            'slot-deployment.class': slot_deployment_klass
+        }
+
+        if any([i for i in [jobId, issuer, ipVersion, slot_deployment_klass]
+                if i is None]):
+            raise ValueError("San info error: %s" % res)
+
+        return res
+
+    def save_cert_last_spsid(self, san_cert_name, sps_id_value):
+        san_info = self._get_akamai_san_certs_info()
+        the_san_cert_info = san_info.get(
+            san_cert_name
+        )
+
+        if the_san_cert_info is None:
+            raise ValueError('No san cert info found for %s.' % san_cert_name)
+
+        the_san_cert_info['spsId'] = sps_id_value
+        san_info[san_cert_name] = the_san_cert_info
+        # Change the previous san info in the overall provider_info dictionary
+        provider_info = dict(self._get_akamai_provider_info()['info'])
+        provider_info['san_info'] = json.dumps(san_info)
+
+        stmt = query.SimpleStatement(
+            UPDATE_PROVIDER_INFO,
+            consistency_level=self.consistency_level)
+
+        args = {
+            'provider_name': 'akamai',
+            'info': provider_info
+        }
+
+        self.session.execute(stmt, args)
+
+    def get_cert_last_spsid(self, san_cert_name):
+        the_san_cert_info = self._get_akamai_san_certs_info().get(
+            san_cert_name
+        )
+
+        if the_san_cert_info is None:
+            raise ValueError('No san cert info found for %s.' % san_cert_name)
+
+        spsId = the_san_cert_info.get('spsId')
+        return spsId
+
+    def update_san_info(self, san_info_dict):
+        provider_info = {}
+        provider_info['san_info'] = json.dumps(san_info_dict)
+
+        stmt = query.SimpleStatement(
+            CREATE_PROVIDER_INFO,
+            consistency_level=self.consistency_level)
+
+        args = {
+            'provider_name': 'akamai',
+            'info': provider_info
+        }
+
+        self.session.execute(stmt, args)

poppy/provider/akamai/services.py

@@ -49,6 +49,10 @@ class ServiceController(base.ServiceBase):
     def mod_san_queue(self):
         return self.driver.mod_san_queue
 
+    @property
+    def san_cert_cnames(self):
+        return self.driver.san_cert_cnames
+
     def __init__(self, driver):
         super(ServiceController, self).__init__(driver)
 
@@ -58,8 +62,6 @@ class ServiceController(base.ServiceBase):
         self.sps_api_base_url = self.driver.akamai_sps_api_base_url
         self.request_header = {'Content-type': 'application/json',
                                'Accept': 'text/plain'}
-
-        self.san_cert_cnames = self.driver.san_cert_cnames
         self.san_cert_hostname_limit = self.driver.san_cert_hostname_limit
 
     def create(self, service_obj):
@@ -496,64 +498,81 @@ class ServiceController(base.ServiceBase):
 
     def create_certificate(self, cert_obj):
         if cert_obj.cert_type == 'san':
-            for san_cert_name in self.san_cert_cnames:
-                lastSpsId = (
-                    self.san_info_storage.get_cert_last_spsid(san_cert_name))
-                if lastSpsId not in [None, ""]:
-                    LOG.info('Latest spsId for %s is: %s' % (san_cert_name,
-                                                             lastSpsId))
-                    resp = self.sps_api_client.get(
-                        self.sps_api_base_url.format(spsId=lastSpsId),
+            try:
+                for san_cert_name in self.san_cert_cnames:
+                    lastSpsId = (
+                        self.san_info_storage.get_cert_last_spsid(
+                            san_cert_name
+                        )
                     )
-                    if resp.status_code != 200:
-                        raise RuntimeError('SPS API Request Failed'
+                    if lastSpsId not in [None, ""]:
+                        LOG.info('Latest spsId for %s is: %s' % (san_cert_name,
+                                                                 lastSpsId))
+                        resp = self.sps_api_client.get(
+                            self.sps_api_base_url.format(spsId=lastSpsId),
+                        )
+                        if resp.status_code != 200:
+                            raise RuntimeError('SPS API Request Failed'
+                                               'Exception: %s' % resp.text)
+                        status = json.loads(resp.text)['requestList'][0][
+                            'status']
+                        # This SAN Cert is on pending status
+                        if status != 'SPS Request Complete':
+                            LOG.info("SPS Not completed for %s..." %
+                                     self.san_cert_name)
+                            continue
+                    # issue modify san_cert sps request
+                    cert_info = self.san_info_storage.get_cert_info(
+                        san_cert_name)
+                    cert_info['add.sans'] = cert_obj.domain_name
+                    string_post_data = '&'.join(
+                        ['%s=%s' % (k, v) for (k, v) in cert_info.items()])
+                    LOG.info('Post modSan request with request data: %s' %
+                             string_post_data)
+                    resp = self.sps_api_client.post(
+                        self.sps_api_base_url.format(spsId=""),
+                        data=string_post_data
+                    )
+                    if resp.status_code != 202:
+                        raise RuntimeError('SPS Request failed.'
                                            'Exception: %s' % resp.text)
-                    status = json.loads(resp.text)['requestList'][0]['status']
-                    # This SAN Cert is on pending status
-                    if status != 'SPS Request Complete':
-                        LOG.info("SPS Not completed for %s..." %
-                                 self.san_cert_name)
-                        continue
-                # issue modify san_cert sps request
-                cert_info = self.san_info_storage.get_cert_info(san_cert_name)
-                cert_info['add.sans'] = cert_obj.domain_name
-                string_post_data = '&'.join(
-                    ['%s=%s' % (k, v) for (k, v) in cert_info.items()])
-                LOG.info('Post modSan request with request data: %s' %
-                         string_post_data)
-                resp = self.sps_api_client.post(
-                    self.sps_api_base_url.format(spsId=""),
-                    data=string_post_data
-                )
-                if resp.status_code != 202:
-                    raise RuntimeError('SPS Request failed.'
-                                       'Exception: %s' % resp.text)
+                    else:
+                        resp_dict = json.loads(resp.text)
+                        LOG.info('modSan request submitted. Response: %s' %
+                                 str(resp_dict))
+                        this_sps_id = resp_dict['spsId']
+                        self.san_info_storage.save_cert_last_spsid(
+                            san_cert_name,
+                            this_sps_id)
+                        return self.responder.ssl_certificate_provisioned(
+                            san_cert_name, {
+                                'status': 'create_in_progress',
+                                'san cert': san_cert_name,
+                                'akamai_spsId': this_sps_id,
+                                'create_at': str(datetime.datetime.now()),
+                                'action': 'Waiting for customer domain '
+                                          'validation for %s' %
+                                          (cert_obj.domain_name)
+                            })
                 else:
-                    resp_dict = json.loads(resp.text)
-                    LOG.info('modSan request submitted. Response: %s' %
-                             str(resp_dict))
-                    this_sps_id = resp_dict['spsId']
-                    self.san_info_storage.save_cert_last_spsid(san_cert_name,
-                                                               this_sps_id)
-                    return self.responder.ssl_certificate_provisioned(
-                        san_cert_name, {
-                            'status': 'create_in_progress',
-                            'san cert': san_cert_name,
-                            'akamai_spsId': this_sps_id,
-                            'create_at': str(datetime.datetime.now()),
-                            'action': 'Waiting for customer domain '
-                                      'validation for %s' %
-                                      (cert_obj.domain_name)
-                        })
-            else:
-                self.mod_san_queue.enqueue_mod_san_request(
-                    json.dumps(cert_obj.to_dict()))
+                    self.mod_san_queue.enqueue_mod_san_request(
+                        json.dumps(cert_obj.to_dict()))
+                    return self.responder.ssl_certificate_provisioned(None, {
+                        'status': 'failed',
+                        'san cert': None,
+                        'action': 'No available san cert for %s right now,'
+                                  ' or no san cert info available.'
+                                  ' More provisioning might be needed' %
+                                  (cert_obj.domain_name)
+                    })
+            except Exception as e:
                 return self.responder.ssl_certificate_provisioned(None, {
                     'status': 'failed',
                     'san cert': None,
-                    'action': 'No available san cert for %s right now.'
-                              ' More provisioning might be needed' %
-                              (cert_obj.domain_name)
+                    'action': 'Waiting for action... '
+                              'Provision san cert failed for %s failed.'
+                              ' Reason: %s' %
+                              (cert_obj.domain_name, str(e))
                 })
         else:
             return self.responder.ssl_certificate_provisioned(None, {

poppy/storage/cassandra/migrations/006_provider_info.cql

@@ -0,0 +1,10 @@
+CREATE TABLE providers_info (
+    provider_name VARCHAR,
+    info MAP<TEXT, TEXT>,
+    PRIMARY KEY (provider_name)
+);
+
+
+--//@UNDO
+
+DROP TABLE IF EXISTS providers_info;

scripts/providers/akamai/san_cert_info/cassandra/list_san_cert_info.py

@@ -0,0 +1,41 @@
+# Copyright (c) 2015 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 poppy.provider.akamai.san_info_storage import cassandra_storage
+
+
+CONF = cfg.CONF
+CONF.register_cli_opts(cassandra_storage.CASSANDRA_OPTIONS,
+                       group=cassandra_storage.AKAMAI_CASSANDRA_STORAGE_GROUP)
+CONF(prog='akamai-config')
+
+
+def main():
+    v_cassandra_storage = cassandra_storage.CassandraSanInfoStorage(CONF)
+
+    all_san_cert_names = v_cassandra_storage.list_all_san_cert_names()
+
+    if not all_san_cert_names:
+        print ("Currently no SAN cert info has been initialized")
+
+    for san_cert_name in all_san_cert_names:
+        print("%s:%s" % (san_cert_name,
+                         str(v_cassandra_storage.get_cert_info(san_cert_name)))
+              )
+
+if __name__ == "__main__":
+    main()

scripts/providers/akamai/san_cert_info/cassandra/upsert_san_cert_info.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2015 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 poppy.provider.akamai import driver
+from poppy.provider.akamai.san_info_storage import cassandra_storage
+
+
+CONF = cfg.CONF
+CONF.register_cli_opts(cassandra_storage.CASSANDRA_OPTIONS,
+                       group=cassandra_storage.AKAMAI_CASSANDRA_STORAGE_GROUP)
+CONF.register_cli_opts(driver.AKAMAI_OPTIONS, driver.AKAMAI_GROUP)
+CONF(prog='akamai-config')
+
+
+def main():
+    v_cassandra_storage = cassandra_storage.CassandraSanInfoStorage(CONF)
+
+    san_attribute_default_list = {
+        'issuer': 'symentec',
+        'ipVersion': 'ipv4',
+        'slot_deployment_klass': 'esslType',
+        'jobId': None}
+
+    san_info_dict = {
+    }
+    for san_cert_name in CONF[driver.AKAMAI_GROUP].san_cert_cnames:
+        san_info_dict[san_cert_name] = {}
+        print("Insert SAN info for :%s" % (san_cert_name))
+        for attr in san_attribute_default_list:
+            user_input = None
+            while ((user_input or "").strip() or user_input) in ["", None]:
+                user_input = raw_input('Please input value for attr: %s, '
+                                       'San cert: %s,'
+                                       'default value: %s'
+                                       ' (if default is None, '
+                                       'that means a real value has to'
+                                       ' be input): ' %
+                                       (attr,
+                                        san_cert_name,
+                                        san_attribute_default_list[attr]))
+                if san_attribute_default_list[attr] is None:
+                    continue
+                else:
+                    user_input = san_attribute_default_list[attr]
+                    break
+            san_info_dict[san_cert_name][attr] = user_input
+
+    v_cassandra_storage.update_san_info(san_info_dict)
+
+
+if __name__ == "__main__":
+    '''
+    example usage:
+    python upsert_san_cert_info.py --config-file ~/.poppy/poppy.conf
+    '''
+    main()

setup.cfg

@@ -63,6 +63,10 @@ poppy.distributed_task =
 poppy.notification =
     mailgun = poppy.notification.mailgun:Driver
 
+poppy.provider.akamai.san_info_storage =
+    zookeeper = poppy.provider.akamai.san_info_storage.zookeeper_storage:ZookeeperSanInfoStorage
+    cassandra = poppy.provider.akamai.san_info_storage.cassandra_storage:CassandraSanInfoStorage
+
 
 
 [wheel]

tests/unit/provider/akamai/san_info_storage/test_cassandra_san_info_storage.py

@@ -0,0 +1,115 @@
+# Copyright (c) 2015 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 json
+
+import ddt
+import mock
+from oslo_config import cfg
+
+from poppy.provider.akamai.san_info_storage import cassandra_storage
+from tests.unit import base
+
+
+@ddt.ddt
+class TestCassandraSANInfoStorage(base.TestCase):
+
+    def setUp(self):
+        super(TestCassandraSANInfoStorage, self).setUp()
+
+        # create mocked config and driver
+        migrations_patcher = mock.patch(
+            'cdeploy.migrator.Migrator'
+        )
+        migrations_patcher.start()
+        self.addCleanup(migrations_patcher.stop)
+
+        cluster_patcher = mock.patch(
+            'cassandra.cluster.Cluster'
+        )
+        cluster_patcher.start()
+        self.addCleanup(cluster_patcher.stop)
+
+        self.conf = cfg.ConfigOpts()
+        self.cassa_storage = cassandra_storage.CassandraSanInfoStorage(
+            self.conf)
+
+        self.get_returned_value = [{'info': {
+            'san_info':
+                '{"secure2.san1.altcdn.com": '
+                '    {"ipVersion": "ipv4", "issuer": "symentec", '
+                '     "slot_deployment_klass": "esslType", "jobId": "4312"},'
+                '"secure1.san1.altcdn.com": '
+                '{"ipVersion": "ipv4", "issuer": "symentec", '
+                '"slot_deployment_klass": "esslType", '
+                '"jobId": "1432", "spsId": 1423}}'}}]
+
+    def test__get_akamai_provider_info(self):
+        mock_execute = self.cassa_storage.session.execute
+        mock_execute.return_value = self.get_returned_value
+        res = self.cassa_storage._get_akamai_provider_info()
+        mock_execute.assert_called()
+        self.assertTrue(res == self.get_returned_value[0])
+
+    def test__get_akamai_san_certs_info(self):
+        mock_execute = self.cassa_storage.session.execute
+        mock_execute.return_value = self.get_returned_value
+
+        res = self.cassa_storage._get_akamai_san_certs_info()
+        mock_execute.assert_called()
+        self.assertTrue(
+            res == json.loads(self.get_returned_value[0]['info']['san_info'])
+        )
+
+    def test_list_all_san_cert_names(self):
+        mock_execute = self.cassa_storage.session.execute
+        mock_execute.return_value = self.get_returned_value
+
+        res = self.cassa_storage.list_all_san_cert_names()
+        mock_execute.assert_called()
+        self.assertTrue(
+            res == json.loads(self.get_returned_value[0]['info']['san_info']).
+            keys()
+        )
+
+    @ddt.data("secure1.san1.altcdn.com", "secure2.san1.altcdn.com")
+    def test_save_cert_last_spsid(self, san_cert_name):
+        mock_execute = self.cassa_storage.session.execute
+        mock_execute.return_value = self.get_returned_value
+
+        self.cassa_storage.save_cert_last_spsid(
+            san_cert_name,
+            '1234'
+        )
+        self.assertTrue(mock_execute.call_count == 3)
+
+    def test_get_cert_last_spsid(self):
+        mock_execute = self.cassa_storage.session.execute
+        mock_execute.return_value = self.get_returned_value
+        cert_name = "secure1.san1.altcdn.com"
+
+        res = self.cassa_storage.get_cert_last_spsid(
+            cert_name
+        )
+        mock_execute.assert_called()
+        self.assertTrue(
+            res == json.loads(self.get_returned_value[0]['info']['san_info'])
+            [cert_name]['spsId']
+        )
+
+    def test_update_san_info(self):
+        mock_execute = self.cassa_storage.session.execute
+        self.cassa_storage.update_san_info({})
+        mock_execute.assert_called()

tests/unit/provider/akamai/test_driver.py

@@ -84,6 +84,9 @@ AKAMAI_OPTIONS = [
     cfg.IntOpt('san_cert_hostname_limit', default=80,
                help='default limit on how many hostnames can'
                ' be held by a SAN cert'),
+    cfg.StrOpt('san_info_storage_type',
+               default='zookeeper',
+               help='Storage type for storing san cert information'),
 
     # related info for SPS && PAPI APIs
     cfg.StrOpt(

tests/unit/provider/akamai/test_services.py

@@ -450,14 +450,15 @@ class TestServices(base.TestCase):
         self.assertTrue(restriction_rule_valid)
 
     def test_create_ssl_certificate_happy_path(self):
+        self.driver.san_cert_cnames = ["secure.san1.poppycdn.com",
+                                       "secure.san2.poppycdn.com"]
+
         controller = services.ServiceController(self.driver)
         data = {
             "cert_type": "san",
             "domain_name": "www.abc.com",
             "flavor_id": "premium"
         }
-        controller.san_cert_cnames = ["secure.san1.poppycdn.com",
-                                      "secure.san2.poppycdn.com"]
 
         lastSpsId = (
             controller.san_info_storage.get_cert_last_spsid(