Rainbond/api/handler/tenant.go

// Copyright (C) 2014-2018 Goodrain Co., Ltd.
// RAINBOND, Application Management Platform

// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version. For any non-GPL usage of Rainbond,
// one or multiple Commercial Licenses authorized by Goodrain Co., Ltd.
// must be obtained first.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

package handler

import (
	"encoding/json"
	"fmt"
	"github.com/goodrain/rainbond/cmd/api/option"
	"net/http"
	"strconv"
	"strings"

	"github.com/Sirupsen/logrus"
	api_model "github.com/goodrain/rainbond/api/model"
	"github.com/goodrain/rainbond/api/util"
	"github.com/goodrain/rainbond/db"
	dbmodel "github.com/goodrain/rainbond/db/model"
	"github.com/goodrain/rainbond/mq/api/grpc/pb"
	cli "github.com/goodrain/rainbond/node/nodem/client"
	"github.com/goodrain/rainbond/worker/client"
)

//TenantAction tenant act
type TenantAction struct {
	MQClient  pb.TaskQueueClient
	statusCli *client.AppRuntimeSyncClient
	OptCfg    *option.Config
}

//CreateTenManager create Manger
func CreateTenManager(MQClient pb.TaskQueueClient, statusCli *client.AppRuntimeSyncClient,
	optCfg *option.Config) *TenantAction {
	return &TenantAction{
		MQClient:  MQClient,
		statusCli: statusCli,
		OptCfg:    optCfg,
	}
}

//GetTenants get tenants
func (t *TenantAction) GetTenants() ([]*dbmodel.Tenants, error) {
	tenants, err := db.GetManager().TenantDao().GetALLTenants()
	if err != nil {
		return nil, err
	}
	return tenants, err
}

//GetTenantsByEid GetTenantsByEid
func (t *TenantAction) GetTenantsByEid(eid string) ([]*dbmodel.Tenants, error) {
	tenants, err := db.GetManager().TenantDao().GetTenantByEid(eid)
	if err != nil {
		return nil, err
	}
	return tenants, err
}

//GetTenantsPaged GetTenantsPaged
func (t *TenantAction) GetTenantsPaged(offset, len int) ([]*dbmodel.Tenants, error) {
	tenants, err := db.GetManager().TenantDao().GetALLTenants()
	if err != nil {
		return nil, err
	}
	return tenants, err
}

//TotalMemCPU StatsMemCPU
func (t *TenantAction) TotalMemCPU(services []*dbmodel.TenantServices) (*api_model.StatsInfo, error) {
	cpus := 0
	mem := 0
	for _, service := range services {
		logrus.Debugf("service is %d, cpus is %d, mem is %v", service.ID, service.ContainerCPU, service.ContainerMemory)
		cpus += service.ContainerCPU
		mem += service.ContainerMemory
	}
	si := &api_model.StatsInfo{
		CPU: cpus,
		MEM: mem,
	}
	return si, nil
}

//GetTenantsName get tenants name
func (t *TenantAction) GetTenantsName() ([]string, error) {
	tenants, err := db.GetManager().TenantDao().GetALLTenants()
	if err != nil {
		return nil, err
	}
	var result []string
	for _, v := range tenants {
		result = append(result, strings.ToLower(v.Name))
	}
	return result, err
}

//GetTenantsByName get tenants
func (t *TenantAction) GetTenantsByName(name string) (*dbmodel.Tenants, error) {
	tenant, err := db.GetManager().TenantDao().GetTenantIDByName(name)
	if err != nil {
		return nil, err
	}

	return tenant, err
}

//GetTenantsByUUID get tenants
func (t *TenantAction) GetTenantsByUUID(uuid string) (*dbmodel.Tenants, error) {
	tenant, err := db.GetManager().TenantDao().GetTenantByUUID(uuid)
	if err != nil {
		return nil, err
	}

	return tenant, err
}

//StatsMemCPU StatsMemCPU
func (t *TenantAction) StatsMemCPU(services []*dbmodel.TenantServices) (*api_model.StatsInfo, error) {
	cpus := 0
	mem := 0
	for _, service := range services {
		status := t.statusCli.GetStatus(service.ServiceID)
		if t.statusCli.IsClosedStatus(status) {
			continue
		}
		cpus += service.ContainerCPU
		mem += service.ContainerMemory
	}
	si := &api_model.StatsInfo{
		CPU: cpus,
		MEM: mem,
	}
	return si, nil
}

// QueryResult contains result data for a query.
type QueryResult struct {
	Data struct {
		Type   string                   `json:"resultType"`
		Result []map[string]interface{} `json:"result"`
	} `json:"data"`
	Status string `json:"status"`
}

//GetTenantsResources Gets the resource usage of the specified tenant.
func (t *TenantAction) GetTenantsResources(tr *api_model.TenantResources) (map[string]map[string]interface{}, error) {
	ids, err := db.GetManager().TenantDao().GetTenantIDsByNames(tr.Body.TenantNames)
	if err != nil {
		return nil, err
	}
	limits, err := db.GetManager().TenantDao().GetTenantLimitsByNames(tr.Body.TenantNames)
	if err != nil {
		return nil, err
	}
	services, err := db.GetManager().TenantServiceDao().GetServicesByTenantIDs(ids)
	if err != nil {
		return nil, err
	}
	// get cluster resources
	allCPU, allMem, err := t.GetAllocatableResources()
	if err != nil {
		return nil, fmt.Errorf("error getting allocatalbe cpu and memory: %v", err)
	}

	var serviceIDs []string
	var serviceMap = make(map[string]dbmodel.TenantServices, len(services))
	var serviceTenantRunning = make(map[string]int, len(ids))
	var serviceTenantCount = make(map[string]int, len(ids))
	serviceStatus := t.statusCli.GetAllStatus()
	for _, s := range services {
		serviceIDs = append(serviceIDs, s.ServiceID)
		serviceMap[s.ServiceID] = *s
		if !t.statusCli.IsClosedStatus(serviceStatus[s.ServiceID]) {
			serviceTenantRunning[s.TenantID]++
		}
		serviceTenantCount[s.TenantID]++
	}
	var allocatedMemory int64
	for _, v := range limits {
		allocatedMemory = allocatedMemory + int64(v)
	}
	allMem = allMem - allocatedMemory
	var result = make(map[string]map[string]interface{}, len(ids))
	for k, v := range limits {
		result[k] = map[string]interface{}{
			"tenant_id":           k,
			"limit_memory":        v,
			"service_running_num": serviceTenantRunning[k],
			"service_total_num":   serviceTenantCount[k],
			"limit_cpu":           v / 4,
			"cpu":                 0,
			"memory":              0,
			"disk":                0,
		}
		if v == 0 {
			result[k]["limit_memory"] = allMem
			if allMem/4 > allCPU {
				result[k]["limit_cpu"] = allCPU
			} else {
				result[k]["limit_cpu"] = allMem / 4
			}
		}
	}

	status := t.statusCli.GetStatuss(strings.Join(serviceIDs, ","))
	for k, v := range status {
		if _, ok := serviceMap[k]; !ok {
			continue
		}
		if _, ok := result[serviceMap[k].TenantID]; !ok {
			result[serviceMap[k].TenantID] = map[string]interface{}{
				"tenant_id":    k,
				"limit_memory": allMem,
				"limit_cpu":    allMem / 4,
				"cpu":          0,
				"memory":       0,
				"disk":         0,
			}
			if allMem/4 > allCPU {
				result[serviceMap[k].TenantID]["limit_cpu"] = allCPU
			}
		}
		if !t.statusCli.IsClosedStatus(v) {
			result[serviceMap[k].TenantID]["cpu"] = result[serviceMap[k].TenantID]["cpu"].(int) +
				(serviceMap[k].ContainerCPU * serviceMap[k].Replicas)
			result[serviceMap[k].TenantID]["memory"] = result[serviceMap[k].TenantID]["memory"].(int) +
				(serviceMap[k].ContainerMemory * serviceMap[k].Replicas)
		}
	}
	//query disk used in prometheus
	pproxy := GetPrometheusProxy()
	query := fmt.Sprintf(`sum(app_resource_appfs{tenant_id=~"%s"}) by(tenant_id)`, strings.Join(ids, "|"))
	query = strings.Replace(query, " ", "%20", -1)
	req, err := http.NewRequest("GET", fmt.Sprintf("http://127.0.0.1:9999/api/v1/query?query=%s", query), nil)
	if err != nil {
		logrus.Error("create request prometheus api error ", err.Error())
		return result, nil
	}
	presult, err := pproxy.Do(req)
	if err != nil {
		logrus.Error("do pproxy request prometheus api error ", err.Error())
		return result, nil
	}
	if presult.Body != nil {
		defer presult.Body.Close()
		if presult.StatusCode != 200 {
			return result, nil
		}
		var qres QueryResult
		err = json.NewDecoder(presult.Body).Decode(&qres)
		if err == nil {
			for _, re := range qres.Data.Result {
				var tenantID string
				var disk int
				if tid, ok := re["metric"].(map[string]interface{}); ok {
					tenantID = tid["tenant_id"].(string)
				}
				if re, ok := (re["value"]).([]interface{}); ok && len(re) == 2 {
					disk, _ = strconv.Atoi(re[1].(string))
				}
				if _, ok := result[tenantID]; ok {
					result[tenantID]["disk"] = disk / 1024
				}
			}
		}
	}
	return result, nil
}

// GetAllocatableResources returns allocatable cpu and memory
func (t *TenantAction) GetAllocatableResources() (int64, int64, error) {
	var allCPU int64 // allocatable CPU
	var allMem int64 // allocatable memory
	nproxy := GetNodeProxy()
	req, err := http.NewRequest("GET", fmt.Sprintf("http://%s/v2/nodes/rule/compute",
		t.OptCfg.NodeAPI), nil)
	if err != nil {
		return 0, 0, fmt.Errorf("error creating http request: %v", err)
	}
	resp, err := nproxy.Do(req)
	if err != nil {
		return 0, 0, fmt.Errorf("error getting cluster resources: %v", err)
	}
	if resp.Body != nil {
		defer resp.Body.Close()
		if resp.StatusCode != 200 {
			return 0, 0, fmt.Errorf("error getting cluster resources: status code: %d; "+
				"response: %v", resp.StatusCode, resp)
		}
		type foo struct {
			List []*cli.HostNode `json:"list"`
		}
		var f foo
		err = json.NewDecoder(resp.Body).Decode(&f)
		if err != nil {
			return 0, 0, fmt.Errorf("error decoding response body: %v", err)
		}

		for _, n := range f.List {
			if n.Status != "running" {
				logrus.Warningf("node %s isn't running, ignore it", n.ID)
				continue
			}
			if k := n.NodeStatus.KubeNode; k != nil {
				s := strings.Replace(k.Status.Allocatable.Cpu().String(), "m", "", -1)
				i, err := strconv.ParseInt(s, 10, 64)
				if err != nil {
					return 0, 0, fmt.Errorf("error converting string to int64: %v", err)
				}
				allCPU = allCPU + i
				allMem = allMem + k.Status.Allocatable.Memory().Value()/(1024*1024)
			}
		}
	}

	return allCPU, allMem, nil
}

//GetServicesResources Gets the resource usage of the specified service.
func (t *TenantAction) GetServicesResources(tr *api_model.ServicesResources) (re map[string]map[string]interface{}, err error) {
	status := t.statusCli.GetStatuss(strings.Join(tr.Body.ServiceIDs, ","))
	var running, closed []string
	for k, v := range status {
		if !t.statusCli.IsClosedStatus(v) {
			running = append(running, k)
		} else {
			closed = append(closed, k)
		}
	}
	resmp, err := db.GetManager().TenantServiceDao().GetServiceMemoryByServiceIDs(running)
	if err != nil {
		return nil, err
	}
	for _, c := range closed {
		resmp[c] = map[string]interface{}{"memory": 0, "cpu": 0}
	}
	re = resmp
	appdisks := t.statusCli.GetAppsDisk(strings.Join(tr.Body.ServiceIDs, ","))
	for serviceID, disk := range appdisks {
		if _, ok := resmp[serviceID]; ok {
			resmp[serviceID]["disk"] = disk / 1024
		} else {
			resmp[serviceID] = make(map[string]interface{})
			resmp[serviceID]["disk"] = disk / 1024
		}
	}
	return resmp, nil
}

//TenantsSum TenantsSum
func (t *TenantAction) TenantsSum() (int, error) {
	s, err := db.GetManager().TenantDao().GetALLTenants()
	if err != nil {
		return 0, err
	}
	return len(s), nil
}

//GetProtocols GetProtocols
func (t *TenantAction) GetProtocols() ([]*dbmodel.RegionProcotols, *util.APIHandleError) {
	rps, err := db.GetManager().RegionProcotolsDao().GetAllSupportProtocol("v2")
	if err != nil {
		return nil, util.CreateAPIHandleErrorFromDBError("get all support protocols", err)
	}
	return rps, nil
}

//TransPlugins TransPlugins
func (t *TenantAction) TransPlugins(tenantID, tenantName, fromTenant string, pluginList []string) *util.APIHandleError {
	tenantInfo, err := db.GetManager().TenantDao().GetTenantIDByName(fromTenant)
	if err != nil {
		return util.CreateAPIHandleErrorFromDBError("get tenant infos", err)
	}
	goodrainID := tenantInfo.UUID
	tx := db.GetManager().Begin()
	for _, p := range pluginList {
		pluginInfo, err := db.GetManager().TenantPluginDao().GetPluginByID(p, goodrainID)
		if err != nil {
			tx.Rollback()
			return util.CreateAPIHandleErrorFromDBError("get plugin infos", err)
		}
		pluginInfo.TenantID = tenantID
		pluginInfo.Domain = tenantName
		pluginInfo.ID = 0
		err = db.GetManager().TenantPluginDaoTransactions(tx).AddModel(pluginInfo)
		if err != nil {
			if !strings.Contains(err.Error(), "is exist") {
				tx.Rollback()
				return util.CreateAPIHandleErrorFromDBError("add plugin Info", err)
			}
		}
	}
	if err := tx.Commit().Error; err != nil {
		return util.CreateAPIHandleErrorFromDBError("trans plugins infos", err)
	}
	return nil
}

func (t *TenantAction) GetServicesStatus(ids string) map[string]string {
	return t.statusCli.GetStatuss(ids)
}

func (t *TenantAction) IsClosedStatus(status string) bool {
	return t.statusCli.IsClosedStatus(status)
}