test/Rainbond
1
0
Fork 0
mirror of https://gitee.com/rainbond/Rainbond.git synced 2024-11-30 10:48:15 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

[FIX] fix the bug of get datacenter config

Browse Source
This commit is contained in:
goodrain 2017-12-20 13:48:49 +08:00
parent 277cfb9aea
commit 6d87712cb7
8 changed files with 253 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
pkg/api/region/node.go
View File

@ -17,29 +17,33 @@ import (
utilhttp "github.com/goodrain/rainbond/pkg/util/http"
)
var nodeServer *RNodeServer
var nodeclient *RNodeClient
//NewNode new node client
func NewNode(nodeAPI string) {
if nodeServer == nil {
nodeServer = &RNodeServer{
if nodeclient == nil {
nodeclient = &RNodeClient{
NodeAPI: nodeAPI,
}
}
}
func GetNode() *RNodeServer {
return nodeServer
func GetNode() *RNodeClient {
return nodeclient
}
type RNodeServer struct {
type RNodeClient struct {
NodeAPI string
}
func (r *RNodeServer) Tasks() TaskInterface {
func (r *RNodeClient) Tasks() TaskInterface {
return &Task{}
}
func (r *RNodeServer) Nodes() NodeInterface {
func (r *RNodeClient) Nodes() NodeInterface {
return &Node{}
}
func (r *RNodeClient) Configs() ConfigsInterface {
return &configs{client: r}
}
type Task struct {
TaskID string `json:"task_id"`
@ -70,9 +74,37 @@ type NodeInterface interface {
Label(label map[string]string)
}
//ConfigsInterface 数据中心配置API
type ConfigsInterface interface {
Get() (*model.GlobalConfig, error)
}
type configs struct {
client *RNodeClient
}
func (c *configs) Get() (*model.GlobalConfig, error) {
body, code, err := nodeclient.Request("/configs/datacenter", "GET", nil)
if err != nil {
return nil, err
}
if code != 200 {
return nil, fmt.Errorf("Get database center configs code %d", code)
}
var res utilhttp.ResponseBody
var gc model.GlobalConfig
res.Bean = &gc
if err := ffjson.Unmarshal(body, &res); err != nil {
return nil, err
}
if gc, ok := res.Bean.(*model.GlobalConfig); ok {
return gc, nil
}
return nil, nil
}
func (t *Node) Label(label map[string]string) {
body, _ := json.Marshal(label)
_, _, err := nodeServer.Request("/nodes/"+t.Id+"/labels", "PUT", body)
_, _, err := nodeclient.Request("/nodes/"+t.Id+"/labels", "PUT", body)
if err != nil {
logrus.Errorf("error details %s", err.Error())
}
@ -80,32 +112,32 @@ func (t *Node) Label(label map[string]string) {
func (t *Node) Add(node *model.APIHostNode) {
body, _ := json.Marshal(node)
_, _, err := nodeServer.Request("/nodes", "POST", body)
_, _, err := nodeclient.Request("/nodes", "POST", body)
if err != nil {
logrus.Errorf("error details %s", err.Error())
}
}
func (t *Node) Delete() {
_, _, err := nodeServer.Request("/nodes/"+t.Id, "DELETE", nil)
_, _, err := nodeclient.Request("/nodes/"+t.Id, "DELETE", nil)
if err != nil {
logrus.Errorf("error details %s", err.Error())
}
}
func (t *Node) Up() {
nodeServer.Request("/nodes/"+t.Id+"/up", "POST", nil)
nodeclient.Request("/nodes/"+t.Id+"/up", "POST", nil)
}
func (t *Node) Down() {
nodeServer.Request("/nodes/"+t.Id+"/down", "POST", nil)
nodeclient.Request("/nodes/"+t.Id+"/down", "POST", nil)
}
func (t *Node) UnSchedulable() {
nodeServer.Request("/nodes/"+t.Id+"/unschedulable", "PUT", nil)
nodeclient.Request("/nodes/"+t.Id+"/unschedulable", "PUT", nil)
}
func (t *Node) ReSchedulable() {
nodeServer.Request("/nodes/"+t.Id+"/reschedulable", "PUT", nil)
nodeclient.Request("/nodes/"+t.Id+"/reschedulable", "PUT", nil)
}
func (t *Node) Get(node string) *Node {
body, _, err := nodeServer.Request("/nodes/"+node, "GET", nil)
body, _, err := nodeclient.Request("/nodes/"+node, "GET", nil)
if err != nil {
logrus.Errorf("error get node %s,details %s", node, err.Error())
return nil
@ -133,7 +165,7 @@ func (t *Node) Get(node string) *Node {
}
func (t *Node) Rule(rule string) []*model.HostNode {
body, _, err := nodeServer.Request("/nodes/rule/"+rule, "GET", nil)
body, _, err := nodeclient.Request("/nodes/rule/"+rule, "GET", nil)
if err != nil {
logrus.Errorf("error get rule %s ,details %s", rule, err.Error())
return nil
@ -158,7 +190,7 @@ func (t *Node) Rule(rule string) []*model.HostNode {
return nodes
}
func (t *Node) List() []*model.HostNode {
body, _, err := nodeServer.Request("/nodes", "GET", nil)
body, _, err := nodeclient.Request("/nodes", "GET", nil)
if err != nil {
logrus.Errorf("error get nodes ,details %s", err.Error())
return nil
@ -185,9 +217,8 @@ func (t *Node) List() []*model.HostNode {
func (t *Task) Get(id string) (*Task, error) {
t.TaskID = id
url := "/tasks/" + id
resp, code, err := nodeServer.Request(url, "GET", nil)
resp, code, err := nodeclient.Request(url, "GET", nil)
if err != nil {
logrus.Errorf("error request url %s,details %s", url, err.Error())
return nil, err
}
if code != 200 {
@ -217,9 +248,8 @@ func (t *Task) Get(id string) (*Task, error) {
//List list all task
func (t *Task) List() ([]*model.Task, error) {
url := "/tasks"
resp, _, err := nodeServer.Request(url, "GET", nil)
resp, _, err := nodeclient.Request(url, "GET", nil)
if err != nil {
logrus.Errorf("error request url %s,details %s", url, err.Error())
return nil, err
}
var rb utilhttp.ResponseBody
@ -246,7 +276,7 @@ func (t *Task) Exec(taskID string, nodes []string) error {
nodesBody.Nodes = nodes
body, _ := json.Marshal(nodesBody)
url := "/tasks/" + taskID + "/exec"
resp, code, err := nodeServer.Request(url, "POST", body)
resp, code, err := nodeclient.Request(url, "POST", body)
if code != 200 {
return fmt.Errorf("exec failure," + string(resp))
}
@ -259,7 +289,7 @@ func (t *Task) Add(task *model.Task) error {
body, _ := json.Marshal(task)
url := "/tasks"
resp, code, err := nodeServer.Request(url, "POST", body)
resp, code, err := nodeclient.Request(url, "POST", body)
if code != 200 {
return fmt.Errorf("add task failure," + string(resp))
}
@ -271,7 +301,7 @@ func (t *Task) Add(task *model.Task) error {
func (t *Task) AddGroup(group *model.TaskGroup) error {
body, _ := json.Marshal(group)
url := "/taskgroups"
resp, code, err := nodeServer.Request(url, "POST", body)
resp, code, err := nodeclient.Request(url, "POST", body)
if code != 200 {
return fmt.Errorf("add taskgroup failure," + string(resp))
}
@ -284,7 +314,7 @@ func (t *Task) AddGroup(group *model.TaskGroup) error {
//Refresh 刷新静态配置
func (t *Task) Refresh() error {
url := "/tasks/taskreload"
_, code, err := nodeServer.Request(url, "PUT", nil)
_, code, err := nodeclient.Request(url, "PUT", nil)
if code != 200 {
return fmt.Errorf("refresh error code,%d", code)
}
@ -304,7 +334,7 @@ func (t *Task) Status() (*TaskStatus, error) {
return HandleTaskStatus(taskId)
}
func HandleTaskStatus(task string) (*TaskStatus, error) {
resp, code, err := nodeServer.Request("/tasks/"+task+"/status", "GET", nil)
resp, code, err := nodeclient.Request("/tasks/"+task+"/status", "GET", nil)
if err != nil {
logrus.Errorf("error execute status Request,details %s", err.Error())
return nil, err
@ -339,7 +369,7 @@ func HandleTaskStatus(task string) (*TaskStatus, error) {
}
//Request Request
func (r *RNodeServer) Request(url, method string, body []byte) ([]byte, int, error) {
func (r *RNodeClient) Request(url, method string, body []byte) ([]byte, int, error) {
//logrus.Infof("requesting url: %s by method :%s,and body is ",r.NodeAPI+url,method,string(body))
request, err := http.NewRequest(method, "http://127.0.0.1:6100/v2"+url, bytes.NewBuffer(body))
if err != nil {

7
pkg/grctl/clients/region.go
View File

@ -23,15 +23,20 @@ import (
"github.com/goodrain/rainbond/pkg/api/region"
)
//RegionClient region api
var RegionClient *region.Region
var NodeClient *region.RNodeServer
//NodeClient node api
var NodeClient *region.RNodeClient
//InitRegionClient init region api client
func InitRegionClient(reg option.RegionAPI) error {
region.NewRegion(reg.URL, reg.Token, reg.Type)
RegionClient = region.GetRegion()
return nil
}
//InitNodeClient init node api client
func InitNodeClient(nodeAPI string) error {
region.NewNode("http://127.0.0.1:6100/v2")
NodeClient = region.GetNode()

3
pkg/grctl/cmd/cmd.go
View File

@ -44,7 +44,8 @@ func GetCmds() []cli.Command {
//task相关命令
cmds = append(cmds, NewCmdTasks())
//cmds = append(cmds, NewCmdAddNode())
//数据中心配置相关命令
cmds = append(cmds, NewCmdConfigs())
//cmds = append(cmds, NewCmdComputeGroup())
cmds = append(cmds, NewCmdInstall())

55
pkg/grctl/cmd/configs.go Normal file
View File

@ -0,0 +1,55 @@
// RAINBOND, Application Management Platform
// Copyright (C) 2014-2017 Goodrain Co., Ltd.
// 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 cmd
import (
"fmt"
"github.com/apcera/termtables"
"github.com/goodrain/rainbond/pkg/grctl/clients"
"github.com/urfave/cli"
)
//NewCmdConfigs 全局配置相关命令
func NewCmdConfigs() cli.Command {
c := cli.Command{
Name: "configs",
Usage: "系统任务相关命令，grctl tasks -h",
Subcommands: []cli.Command{
cli.Command{
Name: "get",
Usage: "get all datacenter configs",
Action: func(c *cli.Context) error {
configs, err := clients.NodeClient.Configs().Get()
if err != nil {
return err
}
taskTable := termtables.CreateTable()
taskTable.AddHeaders("Name", "CNName", "ValueType", "Value")
for _, config := range configs.Configs {
taskTable.AddRow(config.Name, config.CNName, config.ValueType, config.Value)
}
fmt.Println(taskTable.Render())
return nil
},
},
},
}
return c
}

76
pkg/grctl/cmd/tasks.go
View File

@ -27,6 +27,7 @@ import (
"github.com/apcera/termtables"
"github.com/goodrain/rainbond/pkg/grctl/clients"
"github.com/goodrain/rainbond/pkg/node/api/model"
"github.com/gosuri/uitable"
"github.com/urfave/cli"
)
@ -49,35 +50,70 @@ func NewCmdTasks() cli.Command {
cli.Command{
Name: "list",
Usage: "List all task",
Flags: []cli.Flag{
cli.BoolFlag{
Name: "g",
Usage: "show all task longitudinal",
},
},
Action: func(c *cli.Context) error {
tasks, err := clients.NodeClient.Tasks().List()
if err != nil {
logrus.Println("list all task error,", err.Error())
return err
}
if len(tasks) > 0 {
taskTable := termtables.CreateTable()
taskTable.AddHeaders("ID", "GroupID", "DepTask", "Status", "Scheduler")
for _, v := range tasks {
var depstr string
for _, dep := range v.Temp.Depends {
depstr += fmt.Sprintf("%s(%s);", dep.DependTaskID, dep.DetermineStrategy)
}
var status string
for k, v := range v.Status {
status += fmt.Sprintf("%s:%s(%s);", k, v.Status, v.CompleStatus)
}
var scheduler = v.Scheduler.Mode + ";"
if len(v.Scheduler.Status) == 0 {
scheduler += "暂未调度"
} else {
for k, v := range v.Scheduler.Status {
scheduler += fmt.Sprintf("%s:%s(%s);", k, v.Status, v.SchedulerTime.Format(time.RFC3339))
if c.Bool("g") {
for _, v := range tasks {
table := uitable.New()
table.Wrap = true // wrap columns
table.AddRow("ID", v.ID)
table.AddRow("GroupID", v.GroupID)
var depstr string
for _, dep := range v.Temp.Depends {
depstr += fmt.Sprintf("%s(%s)\n", dep.DependTaskID, dep.DetermineStrategy)
}
var status string
for k, v := range v.Status {
status += fmt.Sprintf("%s:%s(%s)\n", k, v.Status, v.CompleStatus)
}
var scheduler = v.Scheduler.Mode + "\n"
if len(v.Scheduler.Status) == 0 {
scheduler += "暂未调度"
} else {
for k, v := range v.Scheduler.Status {
scheduler += fmt.Sprintf("%s:%s(%s)\n", k, v.Status, v.SchedulerTime.Format(time.RFC3339))
}
}
table.AddRow("DepTask", depstr)
table.AddRow("Status", status)
table.AddRow("Scheduler", scheduler)
fmt.Println(table.String())
fmt.Println("--------------------------------------------")
}
taskTable.AddRow(v.ID, v.GroupID, depstr, status, scheduler)
} else {
taskTable := termtables.CreateTable()
taskTable.AddHeaders("ID", "GroupID", "DepTask", "Status", "Scheduler")
for _, v := range tasks {
var depstr string
for _, dep := range v.Temp.Depends {
depstr += fmt.Sprintf("%s(%s);", dep.DependTaskID, dep.DetermineStrategy)
}
var status string
for k, v := range v.Status {
status += fmt.Sprintf("%s:%s(%s);", k, v.Status, v.CompleStatus)
}
var scheduler = v.Scheduler.Mode + ";"
if len(v.Scheduler.Status) == 0 {
scheduler += "暂未调度"
} else {
for k, v := range v.Scheduler.Status {
scheduler += fmt.Sprintf("%s:%s(%s);", k, v.Status, v.SchedulerTime.Format(time.RFC3339))
}
}
taskTable.AddRow(v.ID, v.GroupID, depstr, status, scheduler)
}
fmt.Println(taskTable.Render())
}
fmt.Println(taskTable.Render())
return nil
}
fmt.Println("not found tasks")

63
pkg/node/core/config/taskconfig.go
View File

@ -22,6 +22,7 @@ import (
"context"
"fmt"
"regexp"
"strconv"
"strings"
"github.com/Sirupsen/logrus"
@ -87,7 +88,10 @@ func GetConfigKey(rk string) string {
//ResettingArray 根据实际配置解析数组字符串
func ResettingArray(groupCtx *GroupContext, source []string) ([]string, error) {
for i, s := range source {
sourcecopy := make([]string, len(source))
// 使用copy
copy(sourcecopy, source)
for i, s := range sourcecopy {
resultKey := reg.FindAllString(s, -1)
for _, rk := range resultKey {
key := strings.ToUpper(GetConfigKey(rk))
@ -95,10 +99,10 @@ func ResettingArray(groupCtx *GroupContext, source []string) ([]string, error) {
// return nil, fmt.Errorf("%s Parameter configuration error.please make sure `${XXX}`", s)
// }
value := GetConfig(groupCtx, key)
source[i] = strings.Replace(s, rk, value, -1)
sourcecopy[i] = strings.Replace(s, rk, value, -1)
}
}
return source, nil
return sourcecopy, nil
}
//GetConfig 获取配置信息
@ -106,7 +110,21 @@ func GetConfig(groupCtx *GroupContext, key string) string {
if groupCtx != nil {
value := groupCtx.Get(key)
if value != nil {
return value.(string)
logrus.Debugf("group config get %s:%s", key, value)
switch value.(type) {
case string:
if value.(string) != "" {
return value.(string)
}
case int:
if value.(int) != 0 {
return strconv.Itoa(value.(int))
}
case []string:
if value.([]string) != nil {
return strings.Join(value.([]string), "|")
}
}
}
}
if dataCenterConfig == nil {
@ -114,7 +132,32 @@ func GetConfig(groupCtx *GroupContext, key string) string {
}
cn := dataCenterConfig.GetConfig(key)
if cn != nil && cn.Value != nil {
return cn.Value.(string)
if cn.ValueType == "string" || cn.ValueType == "" {
return cn.Value.(string)
}
if cn.ValueType == "array" {
switch cn.Value.(type) {
case []string:
return strings.Join(cn.Value.([]string), "|")
case []interface{}:
vas := cn.Value.([]interface{})
result := ""
for _, va := range vas {
switch va.(type) {
case string:
result += va.(string) + "|"
case int:
result += strconv.Itoa(va.(int)) + "|"
}
}
if len(result) > 0 {
return result[0 : len(result)-1]
}
}
}
if cn.ValueType == "int" {
return strconv.Itoa(cn.Value.(int))
}
}
logrus.Warnf("can not find config for key %s", key)
return ""
@ -136,7 +179,11 @@ func ResettingString(groupCtx *GroupContext, source string) (string, error) {
//ResettingMap 根据实际配置解析Map字符串
func ResettingMap(groupCtx *GroupContext, source map[string]string) (map[string]string, error) {
for k, s := range source {
sourcecopy := make(map[string]string, len(source))
for k, v := range source {
sourcecopy[k] = v
}
for k, s := range sourcecopy {
resultKey := reg.FindAllString(s, -1)
for _, rk := range resultKey {
key := strings.ToUpper(GetConfigKey(rk))
@ -144,8 +191,8 @@ func ResettingMap(groupCtx *GroupContext, source map[string]string) (map[string]
// return nil, fmt.Errorf("%s Parameter configuration error.please make sure `${XXX}`", s)
// }
value := GetConfig(groupCtx, key)
source[k] = strings.Replace(s, rk, value, -1)
sourcecopy[k] = strings.Replace(s, rk, value, -1)
}
}
return source, nil
return sourcecopy, nil
}

25
pkg/node/core/config/taskconfig_test.go
View File

@ -17,19 +17,14 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package config
import (
"testing"
"github.com/goodrain/rainbond/cmd/node/option"
)
import "testing"
import "fmt"
func TestResettingArray(t *testing.T) {
c := CreateDataCenterConfig(&option.Conf{
ConfigStoragePath: "/rainbond/acp_configs",
})
c := CreateDataCenterConfig()
c.Start()
defer c.Stop()
groupCtx := NewGroupContext()
groupCtx := NewGroupContext("")
groupCtx.Add("SADAS", "Test")
result, err := ResettingArray(groupCtx, []string{"Sdd${sadas}asd", "${MYSQL_HOST}", "12_${MYSQL_PASS}_sd"})
if err != nil {
@ -39,12 +34,10 @@ func TestResettingArray(t *testing.T) {
}
func TestResettingString(t *testing.T) {
c := CreateDataCenterConfig(&option.Conf{
ConfigStoragePath: "/rainbond/acp_configs",
})
c := CreateDataCenterConfig()
c.Start()
defer c.Stop()
groupCtx := NewGroupContext()
groupCtx := NewGroupContext("")
groupCtx.Add("SADAS", "Test")
result, err := ResettingString(nil, "${MYSQL_HOST}Sdd${sadas}asd")
if err != nil {
@ -52,3 +45,9 @@ func TestResettingString(t *testing.T) {
}
t.Log(result)
}
func TestGroupConfig(t *testing.T) {
groupCtx := NewGroupContext("")
v := groupCtx.Get("API")
fmt.Println("asdasd:", v)
}

12
pkg/node/masterserver/task/scheduler.go
View File

@ -145,8 +145,9 @@ func (t *TaskEngine) startScheduler() {
SchedulerTime: time.Now(),
}
task.Status[next.NodeID] = model.TaskStatus{
JobID: next.ID,
Status: "Start",
JobID: next.ID,
Status: "Start",
StartTime: time.Now(),
}
t.UpdateTask(task)
next.Scheduler = &job.Scheduler{
@ -255,9 +256,14 @@ func (t *TaskEngine) PutSchedul(taskID string, nodeID string) (err error) {
//初步判断任务是否能被创建
if oldjob := t.GetJob(hash); oldjob != nil {
if task.RunMode == string(job.OnlyOnce) || task.RunMode == string(job.Cycle) {
if oldjob.Scheduler != nil && oldjob.Scheduler.SchedulerStatus == "Success" {
if oldjob.Scheduler != nil && oldjob.Scheduler.SchedulerStatus == "Waiting" {
return fmt.Errorf("task %s run on node %s job only run mode %s", taskID, nodeID, job.OnlyOnce)
}
if oldjob.Scheduler != nil && oldjob.Scheduler.SchedulerStatus == "Success" {
if oldjob.RunStatus != nil && oldjob.RunStatus.Status == "Success" {
return fmt.Errorf("task %s run on node %s job only run mode %s", taskID, nodeID, job.OnlyOnce)
}
}
}
}
var jb *job.Job