From: David Zafman <dzafman@redhat.com>
Date: Thu, 10 Sep 2015 03:50:30 +0000 (-0700)
Subject: osd: Add config option osd_read_ec_check_for_errors for testing
X-Git-Tag: v10.0.0~31^2
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=07e749624092e0443aeba31683caed9c4c7a32cf;p=ceph.git

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 78d74ccaa3e3..5c7352a5f44a 100644
--- 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 d6b95a5abed6..47e3b2f88773 100644
--- 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;
       }
     }
   }