Merge pull request #10678: jewel: rbd-mirror: include local pool id i…

…n resync throttle unique key

Reviewed-by: Loic Dachary <ldachary@redhat.com>

qa/workunits/rbd/rbd_mirror.sh

@@ -198,4 +198,30 @@ set_pool_mirror_mode ${CLUSTER2} ${POOL} 'pool'
 wait_for_image_present ${CLUSTER1} ${POOL} ${image} 'present'
 wait_for_image_replay_started ${CLUSTER1} ${POOL} ${image}
 
+testlog "TEST: simple image resync"
+request_resync_image ${CLUSTER1} ${POOL} ${image}
+wait_for_image_present ${CLUSTER1} ${POOL} ${image} 'deleted'
+wait_for_image_replay_started ${CLUSTER1} ${POOL} ${image}
+test_status_in_pool_dir ${CLUSTER1} ${POOL} ${image} 'up+replaying' 'master_position'
+compare_images ${POOL} ${image}
+
+testlog "TEST: image resync while replayer is stopped"
+admin_daemon ${CLUSTER1} rbd mirror stop ${POOL}/${image}
+wait_for_image_replay_stopped ${CLUSTER1} ${POOL} ${image}
+request_resync_image ${CLUSTER1} ${POOL} ${image}
+admin_daemon ${CLUSTER1} rbd mirror start ${POOL}/${image}
+wait_for_image_present ${CLUSTER1} ${POOL} ${image} 'deleted'
+admin_daemon ${CLUSTER1} rbd mirror start ${POOL}/${image}
+wait_for_image_replay_started ${CLUSTER1} ${POOL} ${image}
+test_status_in_pool_dir ${CLUSTER1} ${POOL} ${image} 'up+replaying' 'master_position'
+compare_images ${POOL} ${image}
+
+testlog "TEST: request image resync while daemon is offline"
+stop_mirror ${CLUSTER1}
+request_resync_image ${CLUSTER1} ${POOL} ${image}
+start_mirror ${CLUSTER1}
+wait_for_image_replay_started ${CLUSTER1} ${POOL} ${image}
+test_status_in_pool_dir ${CLUSTER1} ${POOL} ${image} 'up+replaying' 'master_position'
+compare_images ${POOL} ${image}
+
 echo OK

qa/workunits/rbd/rbd_mirror_helpers.sh

@@ -408,14 +408,28 @@ wait_for_replay_complete()
     local cluster=$2
     local pool=$3
     local image=$4
-    local s master_pos mirror_pos
-
-    for s in 0.2 0.4 0.8 1.6 2 2 4 4 8 8 16 16 32 32; do
-	sleep ${s}
-	flush "${local_cluster}" "${pool}" "${image}"
-	master_pos=$(get_master_position "${cluster}" "${pool}" "${image}")
-	mirror_pos=$(get_mirror_position "${cluster}" "${pool}" "${image}")
-	test -n "${master_pos}" -a "${master_pos}" = "${mirror_pos}" && return 0
+    local s master_pos mirror_pos last_mirror_pos
+    local master_tag master_entry mirror_tag mirror_entry
+
+    while true; do
+        for s in 0.2 0.4 0.8 1.6 2 2 4 4 8 8 16 16 32 32; do
+	    sleep ${s}
+	    flush "${local_cluster}" "${pool}" "${image}"
+	    master_pos=$(get_master_position "${cluster}" "${pool}" "${image}")
+	    mirror_pos=$(get_mirror_position "${cluster}" "${pool}" "${image}")
+	    test -n "${master_pos}" -a "${master_pos}" = "${mirror_pos}" && return 0
+            test "${mirror_pos}" != "${last_mirror_pos}" && break
+        done
+
+        test "${mirror_pos}" = "${last_mirror_pos}" && return 1
+        last_mirror_pos="${mirror_pos}"
+
+        # handle the case where the mirror is ahead of the master
+        master_tag=$(echo "${master_pos}" | grep -Eo "tag_tid=[0-9]*" | cut -d'=' -f 2)
+        mirror_tag=$(echo "${mirror_pos}" | grep -Eo "tag_tid=[0-9]*" | cut -d'=' -f 2)
+        master_entry=$(echo "${master_pos}" | grep -Eo "entry_tid=[0-9]*" | cut -d'=' -f 2)
+        mirror_entry=$(echo "${mirror_pos}" | grep -Eo "entry_tid=[0-9]*" | cut -d'=' -f 2)
+        test "${master_tag}" = "${mirror_tag}" -a ${master_entry} -le ${mirror_entry} && return 0
     done
     return 1
 }
@@ -660,6 +674,15 @@ wait_for_image_present()
     return 1
 }
 
+request_resync_image()
+{
+    local cluster=$1
+    local pool=$2
+    local image=$3
+
+    rbd --cluster=${cluster} -p ${pool} mirror image resync ${image}
+}
+
 #
 # Main
 #

qa/workunits/rbd/rbd_mirror_stress.sh

@@ -2,6 +2,11 @@
 #
 # rbd_mirror_stress.sh - stress test rbd-mirror daemon
 #
+# The following additional environment variables affect the test:
+#
+#  RBD_MIRROR_REDUCE_WRITES - if not empty, don't run the stress bench write
+#                             tool during the many image test
+#
 
 IMAGE_COUNT=50
 export LOCKDEP=0
@@ -37,25 +42,46 @@ compare_image_snaps()
     rbd --cluster ${CLUSTER2} -p ${pool} export ${image}@${snap_name} ${rmt_export}
     rbd --cluster ${CLUSTER1} -p ${pool} export ${image}@${snap_name} ${loc_export}
     cmp ${rmt_export} ${loc_export}
+    rm -f ${rmt_export} ${loc_export}
 }
 
-wait_for_pool_healthy()
+wait_for_pool_images()
 {
     local cluster=$1
     local pool=$2
     local image_count=$3
     local s
     local count
+    local last_count=0
+
+    while true; do
+        for s in `seq 1 40`; do
+            count=$(rbd --cluster ${cluster} -p ${pool} mirror pool status | grep 'images: ' | cut -d' ' -f 2)
+            test "${count}" = "${image_count}" && return 0
+
+            # reset timeout if making forward progress
+            test $count -gt $last_count && break
+            sleep 30
+        done
+
+        test $count -eq $last_count && return 1
+        $last_count=$count
+    done
+    return 1
+}
+
+wait_for_pool_healthy()
+{
+    local cluster=$1
+    local pool=$2
+    local s
     local state
 
     for s in `seq 1 40`; do
+        state=$(rbd --cluster ${cluster} -p ${pool} mirror pool status | grep 'health:' | cut -d' ' -f 2)
+        test "${state}" = "ERROR" && return 1
+        test "${state}" = "OK" && return 0
 	sleep 30
-        count=$(rbd --cluster ${cluster} -p ${pool} mirror pool status | grep 'images: ')
-        test "${count}" = "images: ${image_count} total" || continue
-
-        state=$(rbd --cluster ${cluster} -p ${pool} mirror pool status | grep 'health:')
-        test "${state}" = "health: ERROR" && return 1
-        test "${state}" = "health: OK" && return 0
     done
     return 1
 }
@@ -90,29 +116,40 @@ remove_image ${CLUSTER2} ${POOL} ${image}
 wait_for_image_present ${CLUSTER1} ${POOL} ${image} 'deleted'
 
 testlog "TEST: create many images"
+snap_name="snap"
 for i in `seq 1 ${IMAGE_COUNT}`
 do
   image="image_${i}"
   create_image ${CLUSTER2} ${POOL} ${image} '128M'
-  write_image ${CLUSTER2} ${POOL} ${image} 100
+  if [ -n "${RBD_MIRROR_REDUCE_WRITES}" ]; then
+    write_image ${CLUSTER2} ${POOL} ${image} 100
+  else
+    stress_write_image ${CLUSTER2} ${POOL} ${image}
+  fi
 done
 
-wait_for_pool_healthy ${CLUSTER2} ${POOL} ${IMAGE_COUNT}
-wait_for_pool_healthy ${CLUSTER1} ${POOL} ${IMAGE_COUNT}
+wait_for_pool_images ${CLUSTER2} ${POOL} ${IMAGE_COUNT}
+wait_for_pool_healthy ${CLUSTER2} ${POOL}
+
+wait_for_pool_images ${CLUSTER1} ${POOL} ${IMAGE_COUNT}
+wait_for_pool_healthy ${CLUSTER1} ${POOL}
 
 testlog "TEST: compare many images"
 for i in `seq 1 ${IMAGE_COUNT}`
 do
   image="image_${i}"
+  create_snap ${CLUSTER2} ${POOL} ${image} ${snap_name}
   wait_for_image_replay_started ${CLUSTER1} ${POOL} ${image}
   wait_for_replay_complete ${CLUSTER1} ${CLUSTER2} ${POOL} ${image}
-  compare_images ${POOL} ${image}
+  wait_for_snap_present ${CLUSTER1} ${POOL} ${image} ${snap_name}
+  compare_image_snaps ${POOL} ${image} ${snap_name}
 done
 
 testlog "TEST: delete many images"
 for i in `seq 1 ${IMAGE_COUNT}`
 do
   image="image_${i}"
+  remove_snapshot ${CLUSTER2} ${POOL} ${image} ${snap_name}
   remove_image ${CLUSTER2} ${POOL} ${image}
 done
 

src/CMakeLists.txt

@@ -1003,6 +1003,7 @@ if(${WITH_RBD})
     tools/rbd_mirror/ImageReplayer.cc
     tools/rbd_mirror/ImageDeleter.cc
     tools/rbd_mirror/ImageSync.cc
+    tools/rbd_mirror/ImageSyncThrottler.cc
     tools/rbd_mirror/Mirror.cc
     tools/rbd_mirror/PoolWatcher.cc
     tools/rbd_mirror/Replayer.cc

src/common/config_opts.h

@@ -1211,6 +1211,12 @@ OPTION(rbd_journal_object_flush_bytes, OPT_INT, 0) // maximum number of pending
 OPTION(rbd_journal_object_flush_age, OPT_DOUBLE, 0) // maximum age (in seconds) for pending commits
 OPTION(rbd_journal_pool, OPT_STR, "") // pool for journal objects
 
+/**
+ * RBD Mirror options
+ */
+OPTION(rbd_mirror_sync_point_update_age, OPT_DOUBLE, 30) // number of seconds between each update of the image sync point object number
+OPTION(rbd_mirror_concurrent_image_syncs, OPT_U32, 5) // maximum number of image syncs in parallel
+
 OPTION(nss_db_path, OPT_STR, "") // path to nss db
 
 

src/journal/JournalMetadata.cc

@@ -587,15 +587,15 @@ void JournalMetadata::get_tags(const boost::optional<uint64_t> &tag_class,
   ctx->send();
 }
 
-void JournalMetadata::add_listener(Listener *listener) {
+void JournalMetadata::add_listener(JournalMetadataListener *listener) {
   Mutex::Locker locker(m_lock);
   while (m_update_notifications > 0) {
     m_update_cond.Wait(m_lock);
   }
   m_listeners.push_back(listener);
 }
 
-void JournalMetadata::remove_listener(Listener *listener) {
+void JournalMetadata::remove_listener(JournalMetadataListener *listener) {
   Mutex::Locker locker(m_lock);
   while (m_update_notifications > 0) {
     m_update_cond.Wait(m_lock);

src/journal/JournalMetadata.h

@@ -13,6 +13,7 @@
 #include "common/WorkQueue.h"
 #include "cls/journal/cls_journal_types.h"
 #include "journal/AsyncOpTracker.h"
+#include "journal/JournalMetadataListener.h"
 #include <boost/intrusive_ptr.hpp>
 #include <boost/noncopyable.hpp>
 #include <boost/optional.hpp>
@@ -41,11 +42,6 @@ class JournalMetadata : public RefCountedObject, boost::noncopyable {
   typedef std::set<Client> RegisteredClients;
   typedef std::list<Tag> Tags;
 
-  struct Listener {
-    virtual ~Listener() {};
-    virtual void handle_update(JournalMetadata *) = 0;
-  };
-
   JournalMetadata(ContextWQ *work_queue, SafeTimer *timer, Mutex *timer_lock,
                   librados::IoCtx &ioctx, const std::string &oid,
                   const std::string &client_id, double commit_interval);
@@ -62,8 +58,8 @@ class JournalMetadata : public RefCountedObject, boost::noncopyable {
   void get_mutable_metadata(uint64_t *minimum_set, uint64_t *active_set,
 			    RegisteredClients *clients, Context *on_finish);
 
-  void add_listener(Listener *listener);
-  void remove_listener(Listener *listener);
+  void add_listener(JournalMetadataListener *listener);
+  void remove_listener(JournalMetadataListener *listener);
 
   void register_client(const bufferlist &data, Context *on_finish);
   void update_client(const bufferlist &data, Context *on_finish);
@@ -150,7 +146,7 @@ class JournalMetadata : public RefCountedObject, boost::noncopyable {
 
 private:
   typedef std::map<uint64_t, uint64_t> AllocatedEntryTids;
-  typedef std::list<Listener*> Listeners;
+  typedef std::list<JournalMetadataListener*> Listeners;
 
   struct CommitEntry {
     uint64_t object_num;

src/journal/JournalMetadataListener.h

@@ -0,0 +1,30 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2016 SUSE LINUX GmbH
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef CEPH_JOURNAL_JOURNAL_METADATA_LISTENER_H
+#define CEPH_JOURNAL_JOURNAL_METADATA_LISTENER_H
+
+namespace journal {
+
+class JournalMetadata;
+
+struct JournalMetadataListener {
+  virtual ~JournalMetadataListener() {};
+  virtual void handle_update(JournalMetadata *) = 0;
+};
+
+} // namespace journal
+
+#endif // CEPH_JOURNAL_JOURNAL_METADATA_LISTENER_H
+

src/journal/JournalRecorder.h

@@ -36,7 +36,7 @@ class JournalRecorder {
 private:
   typedef std::map<uint8_t, ObjectRecorderPtr> ObjectRecorderPtrs;
 
-  struct Listener : public JournalMetadata::Listener {
+  struct Listener : public JournalMetadataListener {
     JournalRecorder *journal_recorder;
 
     Listener(JournalRecorder *_journal_recorder)

src/journal/JournalTrimmer.h

@@ -33,7 +33,7 @@ class JournalTrimmer {
 private:
   typedef std::function<Context*()> CreateContext;
 
-  struct MetadataListener : public JournalMetadata::Listener {
+  struct MetadataListener : public JournalMetadataListener {
     JournalTrimmer *journal_trimmmer;
 
     MetadataListener(JournalTrimmer *journal_trimmmer)

src/journal/Journaler.cc

@@ -244,6 +244,14 @@ void Journaler::flush_commit_position(Context *on_safe) {
   m_metadata->flush_commit_position(on_safe);
 }
 
+void Journaler::add_listener(JournalMetadataListener *listener) {
+  m_metadata->add_listener(listener);
+}
+
+void Journaler::remove_listener(JournalMetadataListener *listener) {
+  m_metadata->remove_listener(listener);
+}
+
 int Journaler::register_client(const bufferlist &data) {
   C_SaferCond cond;
   register_client(data, &cond);

src/journal/Journaler.h

@@ -9,6 +9,7 @@
 #include "include/Context.h"
 #include "include/rados/librados.hpp"
 #include "journal/Future.h"
+#include "journal/JournalMetadataListener.h"
 #include "cls/journal/cls_journal_types.h"
 #include <list>
 #include <map>
@@ -71,6 +72,9 @@ class Journaler {
   void get_mutable_metadata(uint64_t *minimum_set, uint64_t *active_set,
 			    RegisteredClients *clients, Context *on_finish);
 
+  void add_listener(JournalMetadataListener *listener);
+  void remove_listener(JournalMetadataListener *listener);
+
   int register_client(const bufferlist &data);
   void register_client(const bufferlist &data, Context *on_finish);
 

src/journal/Makefile.am

@@ -23,6 +23,7 @@ noinst_HEADERS += \
 	journal/FutureImpl.h \
 	journal/Journaler.h \
 	journal/JournalMetadata.h \
+	journal/JournalMetadataListener.h \
 	journal/JournalPlayer.h \
 	journal/JournalRecorder.h \
 	journal/JournalTrimmer.h \