osd: Add config option osd_read_ec_check_for_errors for testing

All rados suites involving erasure codes should set this option
Fix message when ignoring errors and log to cluster log

Signed-off-by: David Zafman <dzafman@redhat.com>

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 78d74ccaa3e302a6ac3b02d17d7b45c5147d0d2f..5c7352a5f44a547deb9cb0f9c056478e592b09a4 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -617,6 +617,11 @@ OPTION(osd_recover_clone_overlap, OPT_BOOL, true)   // preserve clone_overlap du
 OPTION(osd_op_num_threads_per_shard, OPT_INT, 2)
 OPTION(osd_op_num_shards, OPT_INT, 5)
 
+// Set to true for testing.  Users should NOT set this.
+// If set to true even after reading enough shards to
+// decode the object, any error will be reported.
+OPTION(osd_read_ec_check_for_errors, OPT_BOOL, false) // return error if any ec shard has an error
+
 // Only use clone_overlap for recovery if there are fewer than
 // osd_recover_clone_overlap_limit entries in the overlap set
 OPTION(osd_recover_clone_overlap_limit, OPT_INT, 10)

diff --git a/src/osd/ECBackend.cc b/src/osd/ECBackend.cc
index d6b95a5abed614f841b913360c95093020a72d43..47e3b2f88773f0899fa4e2415fe4cc35c5bdaf7e 100644 (file)
--- a/src/osd/ECBackend.cc
+++ b/src/osd/ECBackend.cc
@@ -1121,11 +1121,21 @@ void ECBackend::handle_sub_read_reply(
          ++is_complete;
        }
       } else {
-       if (!rop.complete[iter->first].errors.empty())
-         dout(10) << __func__ << " Enough copies for " << iter->first << " (ignore errors)" << dendl;
-       ++is_complete;
-       rop.complete[iter->first].errors.clear();
         assert(rop.complete[iter->first].r == 0);
+       if (!rop.complete[iter->first].errors.empty()) {
+         if (cct->_conf->osd_read_ec_check_for_errors) {
+           dout(10) << __func__ << ": Not ignoring errors, use one shard err=" << err << dendl;
+           err = rop.complete[iter->first].errors.begin()->second;
+            rop.complete[iter->first].r = err;
+         } else {
+           get_parent()->clog_error() << __func__ << ": Error(s) ignored for "
+                                      << iter->first << " enough copies available" << "\n";
+           dout(10) << __func__ << " Error(s) ignored for " << iter->first
+                    << " enough copies available" << dendl;
+           rop.complete[iter->first].errors.clear();
+         }
+       }
+       ++is_complete;
       }
     }
   }