// CreateOrUpdateElasticsearchCluster creates an Elasticsearch deployment
CreateOrUpdateElasticsearchCluster creates an Elasticsearch deployment
// CreateOrUpdateElasticsearchCluster creates an Elasticsearch deployment
func CreateOrUpdateElasticsearchCluster(cluster *api.Elasticsearch) error {
	err := isValidConf(cluster)
	if err != nil {
		if wrongConfig {
			return nil
		}
		wrongConfig = true
		return err
	}
	wrongConfig = false
	getNodes(cluster)
	upgradeInProgressNode := getNodeUpgradeInProgress(cluster)
	scheduledUpgradeNodes := getScheduledUpgradeNodes(cluster)
	if upgradeInProgressNode != nil {
		clusterStatus := cluster.Status.DeepCopy()
		index, nodeStatus := getNodeStatus(upgradeInProgressNode.name(), clusterStatus)
		if _, ok := containsNodeTypeInterface(upgradeInProgressNode, scheduledUpgradeNodes); ok {
			logrus.Debugf("Continuing update for %v", upgradeInProgressNode.name())
			upgradeInProgressNode.update(nodeStatus)
		} else {
			logrus.Debugf("Continuing restart for %v", upgradeInProgressNode.name())
			upgradeInProgressNode.restart(nodeStatus)
		}
		nodeState := upgradeInProgressNode.state()
		nodeStatus.UpgradeStatus.ScheduledForUpgrade = nodeState.UpgradeStatus.ScheduledForUpgrade
		nodeStatus.UpgradeStatus.ScheduledForRedeploy = nodeState.UpgradeStatus.ScheduledForRedeploy
		if index == NOT_FOUND_INDEX {
			clusterStatus.Nodes = append(clusterStatus.Nodes, *nodeStatus)
		} else {
			clusterStatus.Nodes[index] = *nodeStatus
		}
		updateNodeStatus(cluster, *clusterStatus)
	} else {
		if len(scheduledUpgradeNodes) > 0 {
			for _, node := range scheduledUpgradeNodes {
				logrus.Debugf("Perform a update for %v", node.name())
				clusterStatus := cluster.Status.DeepCopy()
				index, nodeStatus := getNodeStatus(node.name(), clusterStatus)
				err := node.update(nodeStatus)
				nodeState := node.state()
				nodeStatus.UpgradeStatus.ScheduledForUpgrade = nodeState.UpgradeStatus.ScheduledForUpgrade
				nodeStatus.UpgradeStatus.ScheduledForRedeploy = nodeState.UpgradeStatus.ScheduledForRedeploy
				if index == NOT_FOUND_INDEX {
					clusterStatus.Nodes = append(clusterStatus.Nodes, *nodeStatus)
				} else {
					clusterStatus.Nodes[index] = *nodeStatus
				}
				updateNodeStatus(cluster, *clusterStatus)
				if err != nil {
					logrus.Warnf("Error occurred while updating node %v: %v", node.name(), err)
				}
			}
		} else {
			scheduledRedeployNodes := getScheduledRedeployOnlyNodes(cluster)
			if len(scheduledRedeployNodes) > 0 {
				for _, node := range scheduledRedeployNodes {
					logrus.Debugf("Perform a redeploy for %v", node.name())
					clusterStatus := cluster.Status.DeepCopy()
					index, nodeStatus := getNodeStatus(node.name(), clusterStatus)
					node.restart(nodeStatus)
					nodeState := node.state()
					nodeStatus.UpgradeStatus.ScheduledForUpgrade = nodeState.UpgradeStatus.ScheduledForUpgrade
					nodeStatus.UpgradeStatus.ScheduledForRedeploy = nodeState.UpgradeStatus.ScheduledForRedeploy
					if index == NOT_FOUND_INDEX {
						clusterStatus.Nodes = append(clusterStatus.Nodes, *nodeStatus)
					} else {
						clusterStatus.Nodes[index] = *nodeStatus
					}
					updateNodeStatus(cluster, *clusterStatus)
				}
			} else {
				for _, node := range nodes[nodeMapKey(cluster.Name, cluster.Namespace)] {
					clusterStatus := cluster.Status.DeepCopy()
					index, nodeStatus := getNodeStatus(node.name(), clusterStatus)
					err := isValidConf(cluster)
					if err != nil {
						if wrongConfig {
							return nil
						}
						wrongConfig = true
						return err
					}
					node.create()
					nodeState := node.state()
					nodeStatus.UpgradeStatus.ScheduledForUpgrade = nodeState.UpgradeStatus.ScheduledForUpgrade
					nodeStatus.UpgradeStatus.ScheduledForRedeploy = nodeState.UpgradeStatus.ScheduledForRedeploy
					if index == NOT_FOUND_INDEX {
						nodeStatus.DeploymentName = nodeState.DeploymentName
						nodeStatus.StatefulSetName = nodeState.StatefulSetName
						clusterStatus.Nodes = append(clusterStatus.Nodes, *nodeStatus)
					} else {
						clusterStatus.Nodes[index] = *nodeStatus
					}
					updateNodeStatus(cluster, *clusterStatus)
					updateMinMasters(cluster)
				}
			}
		}
	}
	return UpdateClusterStatus(cluster)
}
func CreateOrUpdateElasticsearchCluster ( cluster * api . Elasticsearch ) error { err := isValidConf ( cluster ) if err != nil { if wrongConfig { return nil } wrongConfig = true return err } wrongConfig = false getNodes ( cluster ) upgradeInProgressNode := getNodeUpgradeInProgress ( cluster ) scheduledUpgradeNodes := getScheduledUpgradeNodes ( cluster ) if upgradeInProgressNode != nil { clusterStatus := cluster . Status . DeepCopy ( ) index , nodeStatus := getNodeStatus ( upgradeInProgressNode . name ( ) , clusterStatus ) if _ , ok := containsNodeTypeInterface ( upgradeInProgressNode , scheduledUpgradeNodes ) ; ok { logrus . Debugf ( " " , upgradeInProgressNode . name ( ) ) upgradeInProgressNode . update ( nodeStatus ) } else { logrus . Debugf ( " " , upgradeInProgressNode . name ( ) ) upgradeInProgressNode . restart ( nodeStatus ) } nodeState := upgradeInProgressNode . state ( ) nodeStatus . UpgradeStatus . ScheduledForUpgrade = nodeState . UpgradeStatus . ScheduledForUpgrade nodeStatus . UpgradeStatus . ScheduledForRedeploy = nodeState . UpgradeStatus . ScheduledForRedeploy if index == NOT_FOUND_INDEX { clusterStatus . Nodes = append ( clusterStatus . Nodes , * nodeStatus ) } else { clusterStatus . Nodes [ index ] = * nodeStatus } updateNodeStatus ( cluster , * clusterStatus ) } else { if len ( scheduledUpgradeNodes ) > 0 { for _ , node := range scheduledUpgradeNodes { logrus . Debugf ( " " , node . name ( ) ) clusterStatus := cluster . Status . DeepCopy ( ) index , nodeStatus := getNodeStatus ( node . name ( ) , clusterStatus ) err := node . update ( nodeStatus ) nodeState := node . state ( ) nodeStatus . UpgradeStatus . ScheduledForUpgrade = nodeState . UpgradeStatus . ScheduledForUpgrade nodeStatus . UpgradeStatus . ScheduledForRedeploy = nodeState . UpgradeStatus . ScheduledForRedeploy if index == NOT_FOUND_INDEX { clusterStatus . Nodes = append ( clusterStatus . Nodes , * nodeStatus ) } else { clusterStatus . Nodes [ index ] = * nodeStatus } updateNodeStatus ( cluster , * clusterStatus ) if err != nil { logrus . Warnf ( " " , node . name ( ) , err ) } } } else { scheduledRedeployNodes := getScheduledRedeployOnlyNodes ( cluster ) if len ( scheduledRedeployNodes ) > 0 { for _ , node := range scheduledRedeployNodes { logrus . Debugf ( " " , node . name ( ) ) clusterStatus := cluster . Status . DeepCopy ( ) index , nodeStatus := getNodeStatus ( node . name ( ) , clusterStatus ) node . restart ( nodeStatus ) nodeState := node . state ( ) nodeStatus . UpgradeStatus . ScheduledForUpgrade = nodeState . UpgradeStatus . ScheduledForUpgrade nodeStatus . UpgradeStatus . ScheduledForRedeploy = nodeState . UpgradeStatus . ScheduledForRedeploy if index == NOT_FOUND_INDEX { clusterStatus . Nodes = append ( clusterStatus . Nodes , * nodeStatus ) } else { clusterStatus . Nodes [ index ] = * nodeStatus } updateNodeStatus ( cluster , * clusterStatus ) } } else { for _ , node := range nodes [ nodeMapKey ( cluster . Name , cluster . Namespace ) ] { clusterStatus := cluster . Status . DeepCopy ( ) index , nodeStatus := getNodeStatus ( node . name ( ) , clusterStatus ) err := isValidConf ( cluster ) if err != nil { if wrongConfig { return nil } wrongConfig = true return err } node . create ( ) nodeState := node . state ( ) nodeStatus . UpgradeStatus . ScheduledForUpgrade = nodeState . UpgradeStatus . ScheduledForUpgrade nodeStatus . UpgradeStatus . ScheduledForRedeploy = nodeState . UpgradeStatus . ScheduledForRedeploy if index == NOT_FOUND_INDEX { nodeStatus . DeploymentName = nodeState . DeploymentName nodeStatus . StatefulSetName = nodeState . StatefulSetName clusterStatus . Nodes = append ( clusterStatus . Nodes , * nodeStatus ) } else { clusterStatus . Nodes [ index ] = * nodeStatus } updateNodeStatus ( cluster , * clusterStatus ) updateMinMasters ( cluster ) } } } } return UpdateClusterStatus ( cluster ) }
CreateOrUpdateElasticsearchCluster creates an Elasticsearch deployment
CreateOrUpdateElasticsearchCluster creates an Elasticsearch deployment
// Verify that we didn't scale up too many masters // if wrongConfig=true then we've already print out error message // don't flood the stderr of the operator with the same message // if there is a node currently being upgraded, work on that first // get all nodes that need only a rollout // TODO: ready cluster for a pod restart first // Verify that we didn't scale up too many masters // if wrongConfig=true then we've already print out error message // don't flood the stderr of the operator with the same message // this is a new status, just append // this is an existing status, update in place // update status here // Scrape cluster health from elasticsearch every time