Merge pull request openshift#131 from openshift-cherrypick-robot/cherry-pick-129-to-master-post-release

[master-post-release] Bug 1703546: Waiting for synced flush

Author: openshift-merge-robot

pkg/k8shandler/deployment.go

@@ -309,17 +309,17 @@ func (node *deploymentNode) restart(upgradeStatus *api.ElasticsearchNodeStatus)
 		}
 
 		if replicas > 0 {
-			if ok, err := DoSynchronizedFlush(node.clusterName, node.self.Namespace); !ok {
-				logrus.Warnf("Unable to perform synchronized flush: %v", err)
-				return
-			}
 
 			// disable shard allocation
 			if ok, err := SetShardAllocation(node.clusterName, node.self.Namespace, api.ShardAllocationNone); !ok {
 				logrus.Warnf("Unable to disable shard allocation: %v", err)
 				return
 			}
 
+			if ok, err := DoSynchronizedFlush(node.clusterName, node.self.Namespace); !ok {
+				logrus.Warnf("Unable to perform synchronized flush: %v", err)
+			}
+
 			// check for available replicas empty
 			// node.self.Status.Replicas
 			// if we aren't at 0, then we need to scale down to 0
@@ -393,17 +393,16 @@ func (node *deploymentNode) update(upgradeStatus *api.ElasticsearchNodeStatus) e
 	if upgradeStatus.UpgradeStatus.UpgradePhase == "" ||
 		upgradeStatus.UpgradeStatus.UpgradePhase == api.ControllerUpdated {
 
-		if ok, err := DoSynchronizedFlush(node.clusterName, node.self.Namespace); !ok {
-			logrus.Warnf("Unable to perform synchronized flush: %v", err)
-			return err
-		}
-
 		// disable shard allocation
 		if ok, err := SetShardAllocation(node.clusterName, node.self.Namespace, api.ShardAllocationNone); !ok {
 			logrus.Warnf("Unable to disable shard allocation: %v", err)
 			return err
 		}
 
+		if ok, err := DoSynchronizedFlush(node.clusterName, node.self.Namespace); !ok {
+			logrus.Warnf("Unable to perform synchronized flush: %v", err)
+		}
+
 		if err := node.executeUpdate(); err != nil {
 			return err
 		}

pkg/k8shandler/elasticsearch.go

@@ -357,7 +357,6 @@ func GetClusterNodeCount(clusterName, namespace string) (int32, error) {
 	return nodeCount, payload.Error
 }
 
-// TODO: also check that the number of shards in the response > 0?
 func DoSynchronizedFlush(clusterName, namespace string) (bool, error) {
 
 	payload := &esCurlStruct{
@@ -367,6 +366,17 @@ func DoSynchronizedFlush(clusterName, namespace string) (bool, error) {
 
 	curlESService(clusterName, namespace, payload)
 
+	failed := 0
+	if shards, ok := payload.ResponseBody["_shards"].(map[string]interface{}); ok {
+		if failedFload, ok := shards["failed"].(float64); ok {
+			failed = int(failedFload)
+		}
+	}
+
+	if payload.Error == nil && failed != 0 {
+		payload.Error = fmt.Errorf("Failed to flush %d shards in preparation for cluster restart", failed)
+	}
+
 	return (payload.StatusCode == 200), payload.Error
 }