Merge pull request #27838 from yuvalif/wip-yuval-pubsub_teuthology-new

rgw/pubsub: add multisite pubsub tests to teuthology

Reviewed-by: Casey Bodley <cbodley@redhat.com>

qa/suites/rgw/multisite/realms/three-zone-plus-pubsub.yaml

@@ -18,3 +18,6 @@ overrides:
             endpoints: [c2.client.0]
           - name: test-zone3
             endpoints: [c1.client.1]
+          - name: test-zone4
+            endpoints: [c2.client.1]
+            is_pubsub: true

qa/tasks/rgw_multisite.py

@@ -11,6 +11,7 @@ from util.rgw import rgwadmin, wait_for_radosgw
 from util.rados import create_ec_pool, create_replicated_pool
 from rgw_multi import multisite
 from rgw_multi.zone_rados import RadosZone as RadosZone
+from rgw_multi.zone_ps import PSZone as PSZone
 
 from teuthology.orchestra import run
 from teuthology import misc
@@ -33,6 +34,7 @@ class RGWMultisite(Task):
 
     * 'is_master' is passed on the command line as --master
     * 'is_default' is passed on the command line as --default
+    * 'is_pubsub' is used to create a zone with tier-type=pubsub
     * 'endpoints' given as client names are replaced with actual endpoints
 
             zonegroups:
@@ -78,6 +80,9 @@ class RGWMultisite(Task):
                   - name: test-zone2
                     is_default: true
                     endpoints: [c2.client.0]
+                  - name: test-zone3
+                    is_pubsub: true
+                    endpoints: [c1.client.1]
 
     """
     def __init__(self, ctx, config):
@@ -369,7 +374,10 @@ def create_zonegroup(cluster, gateways, period, config):
 def create_zone(ctx, cluster, gateways, creds, zonegroup, config):
     """ create a zone with the given configuration """
     zone = multisite.Zone(config['name'], zonegroup, cluster)
-    zone = RadosZone(config['name'], zonegroup, cluster)
+    if config.pop('is_pubsub', False):
+        zone = PSZone(config['name'], zonegroup, cluster)
+    else:
+        zone = RadosZone(config['name'], zonegroup, cluster)
 
     # collect Gateways for the zone's endpoints
     endpoints = config.get('endpoints')

qa/tasks/rgw_multisite_tests.py

@@ -10,10 +10,11 @@ from teuthology.exceptions import ConfigError
 from teuthology.task import Task
 from teuthology import misc
 
-from rgw_multi import multisite, tests
+from rgw_multi import multisite, tests, tests_ps
 
 log = logging.getLogger(__name__)
 
+
 class RGWMultisiteTests(Task):
     """
     Runs the rgw_multi tests against a multisite configuration created by the
@@ -66,6 +67,10 @@ class RGWMultisiteTests(Task):
         result = nose.run(defaultTest=tests.__name__, argv=argv, config=conf)
         if not result:
             raise RuntimeError('rgw multisite test failures')
+        result = nose.run(defaultTest=tests_ps.__name__, argv=argv, config=conf)
+        if not result:
+            raise RuntimeError('rgw multisite pubsub test failures')
+
 
 def get_log_stream():
     """ return a log stream for nose output """
@@ -88,4 +93,5 @@ def get_log_stream():
 
     return LogStream()
 
+
 task = RGWMultisiteTests

src/test/rgw/rgw_multi/tests_ps.py

@@ -7,6 +7,7 @@ from .tests import get_realm, \
     zone_meta_checkpoint, \
     zone_bucket_checkpoint, \
     zone_data_checkpoint, \
+    zonegroup_bucket_checkpoint, \
     check_bucket_eq, \
     gen_bucket_name, \
     get_user, \
@@ -17,7 +18,7 @@ from nose import SkipTest
 from nose.tools import assert_not_equal, assert_equal
 
 # configure logging for the tests module
-log = logging.getLogger('rgw_multi.tests')
+log = logging.getLogger(__name__)
 
 ####################################
 # utility functions for pubsub tests
@@ -129,7 +130,6 @@ NOTIFICATION_SUFFIX = "_notif"
 # pubsub tests
 ##############
 
-
 def test_ps_info():
     """ log information for manual testing """
     return SkipTest("only used in manual testing")
@@ -787,7 +787,7 @@ def test_ps_event_acking():
     parsed_result = json.loads(result)
     for event in parsed_result['events']:
         log.debug('Event (after ack) id: "' + str(event['id']) + '"')
-    assert_equal(len(parsed_result['events']), original_number_of_events - number_of_objects/2)
+    assert len(parsed_result['events']) >= (original_number_of_events - number_of_objects/2)
 
     # cleanup
     sub_conf.del_config()
@@ -893,20 +893,30 @@ def test_ps_versioned_deletion():
     # wait for sync
     zone_bucket_checkpoint(ps_zones[0].zone, zones[0].zone, bucket_name)
 
-    # get the create events from the subscription
+    # get the delete events from the subscription
     result, _ = sub_conf.get_events()
     parsed_result = json.loads(result)
     for event in parsed_result['events']:
         log.debug('Event key: "' + str(event['info']['key']['name']) + '" type: "' + str(event['event']) + '"')
+        assert_equal(str(event['event']), 'OBJECT_DELETE')
 
-    # TODO: verify the specific events
+    # TODO: verify we have exactly 2 events
     assert len(parsed_result['events']) >= 2
 
     # cleanup
+    # follwing is needed for the cleanup in the case of 3-zones
+    # see: http://tracker.ceph.com/issues/39142
+    realm = get_realm()
+    zonegroup = realm.master_zonegroup()
+    zonegroup_conns = ZonegroupConns(zonegroup)
+    try:
+        zonegroup_bucket_checkpoint(zonegroup_conns, bucket_name)
+        zones[0].delete_bucket(bucket_name)
+    except:
+        log.debug('zonegroup_bucket_checkpoint failed, cannot delete bucket')
     sub_conf.del_config()
     notification_conf.del_config()
     topic_conf.del_config()
-    zones[0].delete_bucket(bucket_name)
 
 
 def test_ps_push_http():

src/test/rgw/rgw_multi/zone_ps.py

@@ -6,7 +6,7 @@ import hashlib
 import base64
 import xmltodict
 from time import gmtime, strftime
-from multisite import Zone
+from .multisite import Zone
 import boto3
 from botocore.client import Config
 
@@ -74,6 +74,13 @@ def make_request(conn, method, resource, parameters=None, sign_parameters=False,
     return data, status
 
 
+def print_connection_info(conn):
+    """print info of connection"""
+    print("Host: " + conn.host+':'+str(conn.port))
+    print("AWS Secret Key: " + conn.aws_secret_access_key)
+    print("AWS Access Key: " + conn.aws_access_key_id)
+
+
 class PSTopic:
     """class to set/get/delete a topic
     PUT /topics/<topic name>[?push-endpoint=<endpoint>&[<arg1>=<value1>...]]