func (c CephApi) GetMonitors(mon string, clusterId uuid.UUID, ctxt string) ([]string, error) {
	getMonsRoute := CEPH_API_ROUTES["GetMons"]
	getMonsRoute.Pattern = strings.Replace(getMonsRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)

	resp, err := route_request(getMonsRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return []string{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return []string{}, err
	}
	var monsDet backend.CephMons
	if err := json.Unmarshal(respBody, &monsDet); err != nil {
		return []string{}, err
	}
	if len(monsDet.Mons) > 0 {
		var list []string
		for _, mon := range monsDet.Mons {
			list = append(list, mon.Name)
		}
		return list, nil
	} else {
		return []string{}, errors.New("Couldn't retrieve the mons")
	}
}

func (a *App) FetchClusterDetailsFromProvider(ctxt string, clusterId uuid.UUID) (retVal map[string]map[string]interface{}, err error) {
	retVal = make(map[string]map[string]interface{})
	var result models.RpcResponse
	vars := make(map[string]string)
	vars["cluster-id"] = clusterId.String()
	provider := a.GetProviderFromClusterId(ctxt, clusterId)
	if provider == nil {
		logger.Get().Error("%s-Faield to get provider for cluster: %v", ctxt, clusterId)
		return nil, fmt.Errorf("Faield to get provider for cluster: %v", clusterId)
	}
	err = provider.Client.Call(fmt.Sprintf("%s.%s", provider.Name, "GetClusterSummary"), models.RpcRequest{RpcRequestVars: vars, RpcRequestData: []byte{}, RpcRequestContext: ctxt}, &result)
	if result.Status.StatusCode == http.StatusOK || result.Status.StatusCode == http.StatusPartialContent {
		providerResult := make(map[string]interface{})
		unmarshalError := json.Unmarshal(result.Data.Result, &providerResult)
		if unmarshalError != nil {
			logger.Get().Error("%s - Error unmarshalling the monitoring data from provider %v.Error %v", ctxt, provider.Name, unmarshalError.Error())
			return nil, fmt.Errorf("%s - Error unmarshalling the monitoring data from provider %v.Error %v", ctxt, provider.Name, unmarshalError.Error())
		}
		retVal[provider.Name] = providerResult
	}
	if err != nil {
		return retVal, fmt.Errorf("%s - Call to provider %v failed.Err: %v\n", ctxt, provider.Name, err)
	}
	if result.Status.StatusMessage != "" && (result.Status.StatusCode == http.StatusPartialContent || result.Status.StatusCode == http.StatusInternalServerError || result.Status.StatusCode == http.StatusBadRequest) {
		return retVal, fmt.Errorf("%s - Call to provider %v failed.Err: %v\n", ctxt, provider.Name, result.Status.StatusMessage)
	}
	return retVal, nil
}

func (c CephApi) GetMonStatus(mon string, clusterId uuid.UUID, node string, ctxt string) (backend.MonNodeStatus, error) {
	getMonStatusRoute := CEPH_API_ROUTES["GetMonStatus"]
	getMonStatusRoute.Pattern = strings.Replace(
		getMonStatusRoute.Pattern,
		"{cluster-fsid}",
		clusterId.String(),
		1)
	getMonStatusRoute.Pattern = strings.Replace(
		getMonStatusRoute.Pattern,
		"{mon-name}",
		strings.Split(node, ".")[0],
		1)
	resp, err := route_request(getMonStatusRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return backend.MonNodeStatus{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return backend.MonNodeStatus{}, err
	}
	var monStatus backend.MonNodeStatus
	if err := json.Unmarshal(respBody, &monStatus); err != nil {
		return backend.MonNodeStatus{}, err
	}
	return monStatus, nil
}

func (c CephApi) ExecCmd(mon string, clusterId uuid.UUID, cmd string, ctxt string) (bool, string, error) {
	// Replace cluster id in route pattern
	execCmdRoute := CEPH_API_ROUTES["ExecCmd"]
	execCmdRoute.Pattern = strings.Replace(execCmdRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	command := map[string][]string{"command": strings.Split(cmd, " ")}
	buf, err := json.Marshal(command)
	if err != nil {
		return false, "", errors.New(fmt.Sprintf("Error forming request body. error: %v", err))
	}
	body := bytes.NewBuffer(buf)
	resp, err := route_request(execCmdRoute, mon, body)
	defer closeRespBody(resp)
	if err != nil || resp.StatusCode != http.StatusOK {
		return false, "", errors.New(fmt.Sprintf("Failed to execute command: %s. error: %v", cmd, err))
	} else {
		respBody, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			return false, "", err
		}
		var cmdExecResp models.CephCommandResponse
		if err := json.Unmarshal(respBody, &cmdExecResp); err != nil {
			return false, "", err
		}
		if cmdExecResp.Status != 0 {
			return false, "", fmt.Errorf(cmdExecResp.Error)
		} else {
			return true, cmdExecResp.Out, nil
		}
	}
}

func (c CephApi) CreateCrushNode(mon string, clusterId uuid.UUID, node backend.CrushNodeRequest, ctxt string) (int, error) {
	// Replace cluster id in route pattern
	var cNodeId int
	route := CEPH_API_ROUTES["CreateCrushNode"]
	route.Pattern = strings.Replace(route.Pattern, "{cluster-fsid}", clusterId.String(), 1)

	buf, err := json.Marshal(node)
	if err != nil {
		return cNodeId, errors.New(fmt.Sprintf("Error forming request body. error: %v", err))
	}
	body := bytes.NewBuffer(buf)
	resp, err := route_request(route, mon, body)
	defer closeRespBody(resp)
	if err != nil || (resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted) {
		return cNodeId, errors.New(fmt.Sprintf("Failed to create crush node for cluster: %s. error: %v", clusterId.String(), err))
	}
	ok, err := syncRequestStatus(mon, resp)
	if !ok {
		return cNodeId, err
	}
	cNodes, err := c.GetCrushNodes(mon, clusterId, ctxt)
	if err != nil {
		return cNodeId, err
	}
	for _, cNode := range cNodes {
		if cNode.Name == node.Name {
			return cNode.Id, nil
		}
	}

	return cNodeId, errors.New("Failed to retrieve the Crush Node Id")
}

func (c CephApi) GetClusterNetworks(mon string, clusterId uuid.UUID, ctxt string) (skyringmodels.ClusterNetworks, error) {
	getGetClusterNetworksRoute := CEPH_API_ROUTES["GetClusterNetworks"]
	getGetClusterNetworksRoute.Pattern = strings.Replace(
		getGetClusterNetworksRoute.Pattern,
		"{cluster-fsid}",
		clusterId.String(),
		1)
	resp, err := route_request(getGetClusterNetworksRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return skyringmodels.ClusterNetworks{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return skyringmodels.ClusterNetworks{}, err
	}
	var configs map[string]string = make(map[string]string)
	if err := json.Unmarshal(respBody, &configs); err != nil {
		return skyringmodels.ClusterNetworks{}, err
	}
	var clusterNetworks skyringmodels.ClusterNetworks
	clusterNetworks.Public = configs["public_network"]
	if configs["cluster_network"] == "" {
		clusterNetworks.Cluster = clusterNetworks.Public
	} else {
		clusterNetworks.Cluster = configs["cluster_network"]
	}
	return clusterNetworks, nil
}

func WriteAsyncResponse(taskId uuid.UUID, msg string, result []byte) models.RpcResponse {
	var response models.RpcResponse
	response.Status.StatusCode = http.StatusAccepted
	response.Status.StatusMessage = msg
	response.Data.RequestId = taskId.String()
	response.Data.Result = result
	return response
}

func (c CephApi) GetPGCount(mon string, clusterId uuid.UUID, ctxt string) (map[string]uint64, error) {
	pgStatsRoute := CEPH_API_ROUTES["GetPGCount"]
	pgStatsRoute.Pattern = strings.Replace(pgStatsRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	resp, err := route_request(pgStatsRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	var pgSummary map[string]interface{}
	if err != nil {
		return nil, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal(respBody, &pgSummary); err != nil {
		return nil, err
	}
	pgMap, pgMapOk := pgSummary["pg"].(map[string]interface{})
	if !pgMapOk {
		return nil, fmt.Errorf("%s - Failed to fetch number of pgs for the cluster %v", ctxt, clusterId)
	}

	pgCount := make(map[string]uint64)

	var errPgCountError error
	var warnPgCountError error
	var okPgCountError error

	/*
		Error Pg Count
	*/

	pgCount[skyring_monitoring.CRITICAL], errPgCountError = GetPgStatusBasedCount(monitoring.CRITICAL, clusterId, pgMap)
	if errPgCountError != nil {
		return nil, fmt.Errorf("%s - Err: %v", ctxt, errPgCountError)
	}

	/*
		Warning Pg Count
	*/

	pgCount[skyringmodels.STATUS_WARN], warnPgCountError = GetPgStatusBasedCount(monitoring.WARN, clusterId, pgMap)
	if warnPgCountError != nil {
		return nil, fmt.Errorf("%s - Err: %v", ctxt, warnPgCountError)
	}

	/*
		Clean Pg Count
	*/

	pgCount[skyringmodels.STATUS_OK], okPgCountError = GetPgStatusBasedCount(monitoring.OK, clusterId, pgMap)
	if okPgCountError != nil {
		return nil, fmt.Errorf("%s - Err: %v", ctxt, okPgCountError)
	}

	return pgCount, nil
}

func (s Salt) CreateCluster(clusterName string, fsid uuid.UUID, mons []backend.Mon, ctxt string) (bool, error) {
	mutex.Lock()
	defer mutex.Unlock()
	pyobj, err := pyFuncs["CreateCluster"].Call(clusterName, fsid.String(), mons, ctxt)
	if err == nil {
		return gopy.Bool(pyobj), nil
	}

	return false, err
}

// initFormatXL - save XL format configuration on all disks.
func initFormatXL(storageDisks []StorageAPI) (err error) {
	var (
		jbod             = make([]string, len(storageDisks))
		formats          = make([]*formatConfigV1, len(storageDisks))
		saveFormatErrCnt = 0
	)
	for index, disk := range storageDisks {
		if err = disk.MakeVol(minioMetaBucket); err != nil {
			if err != errVolumeExists {
				saveFormatErrCnt++
				// Check for write quorum.
				if saveFormatErrCnt <= len(storageDisks)-(len(storageDisks)/2+3) {
					continue
				}
				return errXLWriteQuorum
			}
		}
		var u *uuid.UUID
		u, err = uuid.New()
		if err != nil {
			saveFormatErrCnt++
			// Check for write quorum.
			if saveFormatErrCnt <= len(storageDisks)-(len(storageDisks)/2+3) {
				continue
			}
			return err
		}
		formats[index] = &formatConfigV1{
			Version: "1",
			Format:  "xl",
			XL: &xlFormat{
				Version: "1",
				Disk:    u.String(),
			},
		}
		jbod[index] = formats[index].XL.Disk
	}
	for index, disk := range storageDisks {
		formats[index].XL.JBOD = jbod
		formatBytes, err := json.Marshal(formats[index])
		if err != nil {
			return err
		}
		if err = disk.AppendFile(minioMetaBucket, formatConfigFile, formatBytes); err != nil {
			return err
		}
	}
	return nil
}

func (c CephApi) GetCrushNodes(mon string, clusterId uuid.UUID, ctxt string) ([]backend.CrushNode, error) {
	route := CEPH_API_ROUTES["GetCrushNodes"]
	route.Pattern = strings.Replace(route.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	resp, err := route_request(route, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return []backend.CrushNode{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return []backend.CrushNode{}, err
	}
	var nodes []backend.CrushNode
	if err := json.Unmarshal(respBody, &nodes); err != nil {
		return []backend.CrushNode{}, err
	}
	return nodes, nil
}

func (c CephApi) GetOSDs(mon string, clusterId uuid.UUID, ctxt string) ([]backend.CephOSD, error) {
	// Replace cluster id in route pattern
	getOsdsRoute := CEPH_API_ROUTES["GetOSDs"]
	getOsdsRoute.Pattern = strings.Replace(getOsdsRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	resp, err := route_request(getOsdsRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return []backend.CephOSD{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return []backend.CephOSD{}, err
	}
	var osds []backend.CephOSD
	if err := json.Unmarshal(respBody, &osds); err != nil {
		return []backend.CephOSD{}, err
	}
	return osds, nil
}

func (c CephApi) GetClusterStatus(mon string, clusterId uuid.UUID, clusterName string, ctxt string) (status string, err error) {
	// Replace cluster id in route pattern
	getPoolsRoute := CEPH_API_ROUTES["GetClusterStatus"]
	getPoolsRoute.Pattern = strings.Replace(getPoolsRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	resp, err := route_request(getPoolsRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return "", err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return "", err
	}
	var clusterHealth backend.CephClusterHealth
	if err := json.Unmarshal(respBody, &clusterHealth); err != nil {
		return "", err
	}
	return clusterHealth.OverallStatus, nil
}

func lockNode(ctxt string, nodeId uuid.UUID, hostname string, operation string) (*lock.AppLock, error) {
	//lock the node
	locks := make(map[uuid.UUID]string)
	if nodeId.IsZero() {
		//Generate temporary UUID from hostname for the node
		//for locking as the UUID is not available at this point
		id, err := uuid.Parse(util.Md5FromString(hostname))
		if err != nil {
			return nil, fmt.Errorf("Unable to create the UUID for locking for host: %s. error: %v", hostname, err)
		}
		nodeId = *id
	}
	locks[nodeId] = fmt.Sprintf("%s : %s", operation, hostname)
	appLock := lock.NewAppLock(locks)
	if err := GetApp().GetLockManager().AcquireLock(ctxt, *appLock); err != nil {
		return nil, err
	}
	return appLock, nil
}

func (c CephApi) UpdateOSD(mon string, clusterId uuid.UUID, osdId string, params map[string]interface{}, ctxt string) (bool, error) {
	// Replace cluster id in route pattern
	updateOsdRoute := CEPH_API_ROUTES["UpdateOSD"]
	updateOsdRoute.Pattern = strings.Replace(updateOsdRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	updateOsdRoute.Pattern = strings.Replace(updateOsdRoute.Pattern, "{osd-id}", osdId, 1)

	buf, err := json.Marshal(params)
	if err != nil {
		return false, errors.New(fmt.Sprintf("Error forming request body: %v", err))
	}
	body := bytes.NewBuffer(buf)
	resp, err := route_request(updateOsdRoute, mon, body)
	defer closeRespBody(resp)
	if err != nil || (resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted) {
		return false, errors.New(fmt.Sprintf("Failed to update osd-id: %s for cluster: %v.error: %v", osdId, clusterId, err))
	} else {
		ok, err := syncRequestStatus(mon, resp)
		return ok, err
	}
}

func (c CephApi) GetPool(mon string, clusterId uuid.UUID, pool_id int, ctxt string) (backend.CephPool, error) {
	getPoolRoute := CEPH_API_ROUTES["GetPool"]
	getPoolRoute.Pattern = strings.Replace(getPoolRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	poolId := strconv.Itoa(pool_id)
	getPoolRoute.Pattern = strings.Replace(getPoolRoute.Pattern, "{pool-id}", poolId, 1)
	resp, err := route_request(getPoolRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return backend.CephPool{}, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return backend.CephPool{}, err
	}
	var pool backend.CephPool
	if err := json.Unmarshal(respBody, &pool); err != nil {
		return backend.CephPool{}, err
	}
	return pool, nil
}

func (c CephApi) PatchCrushRule(mon string, clusterId uuid.UUID, crushRuleId int, params map[string]interface{}, ctxt string) (bool, error) {
	// Replace cluster id in route pattern
	route := CEPH_API_ROUTES["PatchCrushRule"]
	route.Pattern = strings.Replace(route.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	route.Pattern = strings.Replace(route.Pattern, "{crush-rule-id}", strconv.Itoa(crushRuleId), 1)

	buf, err := json.Marshal(params)
	if err != nil {
		return false, errors.New(fmt.Sprintf("Error forming request body: %v", err))
	}
	body := bytes.NewBuffer(buf)
	resp, err := route_request(route, mon, body)
	defer closeRespBody(resp)
	if err != nil || (resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted) {
		return false, errors.New(fmt.Sprintf("Failed to update crush Rule: %v for cluster: %v.error: %v", crushRuleId, clusterId, err))
	} else {
		ok, err := syncRequestStatus(mon, resp)
		return ok, err
	}
}

func (c CephApi) GetPGSummary(mon string, clusterId uuid.UUID, ctxt string) (backend.PgSummary, error) {

	// Replace cluster id in route pattern
	pgStatsRoute := CEPH_API_ROUTES["PGStatistics"]
	pgStatsRoute.Pattern = strings.Replace(pgStatsRoute.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	resp, err := route_request(pgStatsRoute, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	var pgsummary backend.PgSummary
	if err != nil {
		return pgsummary, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return pgsummary, err
	}
	if err := json.Unmarshal(respBody, &pgsummary); err != nil {
		return pgsummary, err
	}
	return pgsummary, err
}

func (c CephApi) GetCrushRule(mon string, clusterId uuid.UUID, crushRuleId int, ctxt string) (map[string]interface{}, error) {
	// Replace cluster id in route pattern
	route := CEPH_API_ROUTES["GetCrushRule"]
	route.Pattern = strings.Replace(route.Pattern, "{cluster-fsid}", clusterId.String(), 1)
	route.Pattern = strings.Replace(route.Pattern, "{crush-rule-id}", strconv.Itoa(crushRuleId), 1)

	var m map[string]interface{}
	resp, err := route_request(route, mon, bytes.NewBuffer([]byte{}))
	defer closeRespBody(resp)
	if err != nil {
		return m, err
	}
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return m, err
	}
	if err := json.Unmarshal(respBody, &m); err != nil {
		return m, err
	}
	return m, nil
}

func (c CephApi) CreateCrushRule(mon string, clusterId uuid.UUID, rule backend.CrushRuleRequest, ctxt string) (int, error) {
	// Replace cluster id in route pattern
	var cRuleId int
	route := CEPH_API_ROUTES["CreateCrushRule"]
	route.Pattern = strings.Replace(route.Pattern, "{cluster-fsid}", clusterId.String(), 1)

	buf, err := json.Marshal(rule)
	if err != nil {
		return cRuleId, errors.New(fmt.Sprintf("Error forming request body. error: %v", err))
	}
	body := bytes.NewBuffer(buf)
	resp, err := route_request(route, mon, body)
	defer closeRespBody(resp)
	if err != nil || (resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted) {
		return cRuleId, errors.New(fmt.Sprintf("Failed to create crush rule for cluster: %s. error: %v", clusterId.String(), err))
	}

	ok, err := syncRequestStatus(mon, resp)
	if !ok {
		return cRuleId, err
	}
	cRules, err := c.GetCrushRules(mon, clusterId, ctxt)
	if err != nil {
		return cRuleId, err
	}
	for _, cRule := range cRules {
		if val, ok := cRule["name"]; ok {
			if rule.Name == val.(string) {
				if val, ok := cRule["ruleset"]; ok {
					cRuleId = int(val.(float64))
					return cRuleId, nil
				}
			}

		}
	}

	return cRuleId, errors.New("Failed to retrieve the Crush Ruleset Id")
}