From 9a651dad3f9a2fb33d8fe1c41f3f09b940bd8715 Mon Sep 17 00:00:00 2001 From: rainzm Date: Mon, 26 Apr 2021 19:25:18 +0800 Subject: [PATCH 1/3] feat(region): return forwardDetails when query server sshable --- pkg/apis/compute/guest_sshable.go | 7 +++++++ pkg/compute/models/guest_sshable.go | 4 ++++ 2 files changed, 11 insertions(+) diff --git a/pkg/apis/compute/guest_sshable.go b/pkg/apis/compute/guest_sshable.go index dc07184e17..d54d21fbe2 100644 --- a/pkg/apis/compute/guest_sshable.go +++ b/pkg/apis/compute/guest_sshable.go @@ -28,6 +28,13 @@ type GuestSshableMethodData struct { Sshable bool Reason string + + ForwardDetails ForwardDetails +} + +type ForwardDetails struct { + ProxyAgentId string + ProxyEndpointId string } type GuestSshableOutput struct { diff --git a/pkg/compute/models/guest_sshable.go b/pkg/compute/models/guest_sshable.go index 3c51186749..a2cd23d04b 100644 --- a/pkg/compute/models/guest_sshable.go +++ b/pkg/compute/models/guest_sshable.go @@ -283,6 +283,10 @@ func (guest *SGuest) sshableTryForward( Method: compute_api.MethodProxyForward, Host: fwd.BindAddr, Port: fwd.BindPort, + ForwardDetails: compute_api.ForwardDetails{ + ProxyAgentId: fwd.ProxyAgentId, + ProxyEndpointId: fwd.ProxyEndpointId, + }, } return guest.sshableTry( ctx, tryData, methodData, From 67d7c571f29a034b49ccfd416476cb979c3162cd Mon Sep 17 00:00:00 2001 From: rainzm Date: Mon, 26 Apr 2021 19:26:04 +0800 Subject: [PATCH 2/3] fix(ansibleserver): be compatible with empty params --- pkg/ansibleserver/models/ansibleplaybook_instance.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/ansibleserver/models/ansibleplaybook_instance.go b/pkg/ansibleserver/models/ansibleplaybook_instance.go index 7dce2be710..dd665d8462 100644 --- a/pkg/ansibleserver/models/ansibleplaybook_instance.go +++ b/pkg/ansibleserver/models/ansibleplaybook_instance.go @@ -139,6 +139,9 @@ func (ai *SAnsiblePlaybookInstance) runPlaybook(ctx context.Context, userCred mc } convertJO := func(o jsonutils.JSONObject) map[string]interface{} { + if o == nil { + return map[string]interface{}{} + } ret := make(map[string]interface{}) o.Unmarshal(&ret) return ret From fd458a6030e4a1f1944a7c9951f2fbbd056ff990 Mon Sep 17 00:00:00 2001 From: rainzm Date: Mon, 26 Apr 2021 19:27:28 +0800 Subject: [PATCH 3/3] feat(devtool): makes the process of installing the agent more rigorous 1. Use the method returned by server sshable to execute ansible playbook 2. Use direct connection and cloud proxy to find the url that the remote machine can send data back to local influxdb --- pkg/apis/devtool/script.go | 14 +- pkg/apis/devtool/script_const.go | 4 + pkg/devtool/models/script.go | 203 +------------ pkg/devtool/models/script_apply.go | 23 +- pkg/devtool/models/script_apply_record.go | 10 +- pkg/devtool/tasks/apply_script_task.go | 178 +++++++---- pkg/devtool/utils/arg_generator.go | 36 +++ pkg/devtool/utils/doc.go | 15 + pkg/devtool/utils/influxdb_url.go | 353 ++++++++++++++++++++++ 9 files changed, 559 insertions(+), 277 deletions(-) create mode 100644 pkg/devtool/utils/arg_generator.go create mode 100644 pkg/devtool/utils/doc.go create mode 100644 pkg/devtool/utils/influxdb_url.go diff --git a/pkg/apis/devtool/script.go b/pkg/apis/devtool/script.go index 01af4f2507..dd771bff88 100644 --- a/pkg/apis/devtool/script.go +++ b/pkg/apis/devtool/script.go @@ -21,14 +21,6 @@ type ScriptApplyInput struct { // required: true // example: b48c5c84-9952-4394-8ca9-c3b84e946a03 ServerID string - // description: whether to use eip first - // example: true - EipFirst bool - // description: Id of proxyEndpoint - // example: cf1d1a0f-9b9d-4629-8036-af3ed87c0821 - ProxyEndpointId string - // description: whether to automatically select proxy endpoint - AutoChooseProxyEndpoint bool } type ScriptApplyOutput struct { @@ -77,8 +69,6 @@ type ScriptDetails struct { } type SApplyInfo struct { - ServerId string - EipFirst bool - ProxyEndpointId string - TryTimes int + ServerId string + TryTimes int } diff --git a/pkg/apis/devtool/script_const.go b/pkg/apis/devtool/script_const.go index 631e8c001e..0786fb67ca 100644 --- a/pkg/apis/devtool/script_const.go +++ b/pkg/apis/devtool/script_const.go @@ -23,6 +23,10 @@ const ( SCRIPT_APPLY_RECORD_SUCCEED = "succeed" SCRIPT_APPLY_RECORD_FAILED = "failed" + SCRIPT_APPLY_RECORD_FAILCODE_SSHABLE = "ServerNotSshable" + SCRIPT_APPLY_RECORD_FAILCODE_INFLUXDB = "NoReachInfluxdb" + SCRIPT_APPLY_RECORD_FAILCODE_OTHERS = "Others" + SCRIPT_NAME = "monitor agent" SERVICE_TYPE = "devtool" diff --git a/pkg/devtool/models/script.go b/pkg/devtool/models/script.go index e433bfc690..9ecd4641c3 100644 --- a/pkg/devtool/models/script.go +++ b/pkg/devtool/models/script.go @@ -16,26 +16,18 @@ package models import ( "context" - "fmt" - "net/url" - "sync" - - "github.com/coredns/coredns/plugin/pkg/log" "yunion.io/x/jsonutils" + "yunion.io/x/log" "yunion.io/x/pkg/errors" - "yunion.io/x/pkg/util/sets" - proxy_api "yunion.io/x/onecloud/pkg/apis/cloudproxy" comapi "yunion.io/x/onecloud/pkg/apis/compute" api "yunion.io/x/onecloud/pkg/apis/devtool" "yunion.io/x/onecloud/pkg/cloudcommon/db" - "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/devtool/utils" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/modules" - "yunion.io/x/onecloud/pkg/mcclient/modules/cloudproxy" - "yunion.io/x/onecloud/pkg/util/httputils" "yunion.io/x/onecloud/pkg/util/stringutils2" ) @@ -63,110 +55,7 @@ func init() { ), } ScriptManager.SetVirtualObject(ScriptManager) - registerArgGenerator(MonitorAgent, getArgs) -} - -type argGenerator func(ctx context.Context, input api.ScriptApplyInput, details *comapi.ServerDetails) (map[string]interface{}, error) - -var argGenerators = &sync.Map{} - -func registerArgGenerator(name string, ag argGenerator) { - argGenerators.Store(name, ag) -} - -func getArgGenerator(name string) (argGenerator, bool) { - v, ok := argGenerators.Load(name) - if !ok { - return nil, ok - } - return v.(argGenerator), ok -} - -func convertInfluxdbUrl(ctx context.Context, pUrl string, endpointId string) (string, error) { - session := auth.AdminSessionWithInternal(ctx, "", "", "") - filter := jsonutils.NewDict() - filter.Set("proxy_endpoint_id", jsonutils.NewString(endpointId)) - filter.Set("opaque", jsonutils.NewString(pUrl)) - filter.Set("scope", jsonutils.NewString("system")) - lr, err := cloudproxy.Forwards.List(session, filter) - if err != nil { - return "", errors.Wrap(err, "failed to list forward") - } - var port int64 - if len(lr.Data) > 0 { - port, _ = lr.Data[0].Int("bind_port") - } else { - rUrl, err := url.Parse(pUrl) - if err != nil { - return "", errors.Wrap(err, "invalid influxdbUrl?") - } - // create one - createP := jsonutils.NewDict() - createP.Set("proxy_endpoint", jsonutils.NewString(endpointId)) - createP.Set("type", jsonutils.NewString(proxy_api.FORWARD_TYPE_REMOTE)) - createP.Set("remote_addr", jsonutils.NewString(rUrl.Hostname())) - createP.Set("remote_port", jsonutils.NewString(rUrl.Port())) - createP.Set("generate_name", jsonutils.NewString("influxdb proxy")) - createP.Set("opaque", jsonutils.NewString(pUrl)) - forward, err := cloudproxy.Forwards.Create(session, createP) - if err != nil { - return "", errors.Wrapf(err, "unable to create forward with create params %s", createP.String()) - } - port, _ = forward.Int("bind_port") - } - // fetch proxy_endpoint address - ep, err := cloudproxy.ProxyEndpoints.Get(session, endpointId, nil) - if err != nil { - return "", errors.Wrapf(err, "unable to get proxy endpoint %s", endpointId) - } - address, _ := ep.GetString("intranet_ip_addr") - return fmt.Sprintf("https://%s:%d", address, port), nil -} - -func getArgs(ctx context.Context, input api.ScriptApplyInput, detail *comapi.ServerDetails) (map[string]interface{}, error) { - influxdbUrl, err := getInfluxdbUrl(ctx) - if err != nil { - return nil, errors.Wrap(err, "unable to get influxdbUrl") - } - // convert influxdbUrl - if len(input.ProxyEndpointId) > 0 { - influxdbUrl, err = convertInfluxdbUrl(ctx, influxdbUrl, input.ProxyEndpointId) - if err != nil { - return nil, errors.Wrapf(err, "unable to convertInfluxdbUrl %s", influxdbUrl) - } - } - vmId := detail.Id - tenantId := detail.ProjectId - domainId := detail.DomainId - ret := map[string]interface{}{ - "influxdb_url": influxdbUrl, - "influxdb_name": "telegraf", - "onecloud_vm_id": vmId, - "onecloud_tenant_id": tenantId, - "onecloud_domain_id": domainId, - } - return ret, nil -} - -var influxdbUrl string - -func getInfluxdbUrl(ctx context.Context) (string, error) { - if len(influxdbUrl) > 0 { - return influxdbUrl, nil - } - session := auth.GetAdminSession(ctx, "", "") - params := jsonutils.NewDict() - params.Set("interface", jsonutils.NewString("public")) - params.Set("service", jsonutils.NewString("influxdb")) - ret, err := modules.EndpointsV3.List(session, params) - if err != nil { - return "", err - } - if len(ret.Data) == 0 { - return "", fmt.Errorf("no sucn endpoint with 'internal' interface and 'influxdb' service") - } - url, _ := ret.Data[0].GetString("url") - return url, nil + utils.RegisterArgGenerator(MonitorAgent, utils.GetArgs) } var MonitorAgent = "monitor agent" @@ -236,8 +125,6 @@ func (s *SScript) ApplyInfos() ([]api.SApplyInfo, error) { ai := make([]api.SApplyInfo, len(sa)) for i := range ai { ai[i].ServerId = sa[i].GuestId - ai[i].EipFirst = sa[i].EipFirst.Bool() - ai[i].ProxyEndpointId = sa[i].ProxyEndpointId ai[i].TryTimes = sa[i].TryTimes } return ai, nil @@ -249,58 +136,17 @@ func (s *SScript) AllowPerformApply(ctx context.Context, userCred mcclient.Token func (s *SScript) PerformApply(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.ScriptApplyInput) (api.ScriptApplyOutput, error) { output := api.ScriptApplyOutput{} - serverInfo, err := s.checkServer(ctx, userCred, input.ServerID) - if err != nil { - return output, err + var argsGenerator string + if s.Name == MonitorAgent { + argsGenerator = MonitorAgent } - // select proxyEndpoint automatically - if len(input.ProxyEndpointId) == 0 && input.AutoChooseProxyEndpoint { - var proxyEndpointId string - // find suitable proxyEndpoint - // network first - session := auth.GetAdminSession(ctx, "", "") - for _, netId := range serverInfo.NetworkIds { - filter := jsonutils.NewDict() - filter.Set("network_id", jsonutils.NewString(netId)) - lr, err := cloudproxy.ProxyEndpoints.List(session, filter) - if err != nil { - return output, errors.Wrapf(err, "unable to list proxy endpoint in network %q", netId) - } - if len(lr.Data) == 0 { - continue - } - proxyEndpointId, _ = lr.Data[0].GetString("id") - break - } - if len(proxyEndpointId) == 0 { - filter := jsonutils.NewDict() - filter.Set("vpc_id", jsonutils.NewString(serverInfo.VpcId)) - lr, err := cloudproxy.ProxyEndpoints.List(session, filter) - if err != nil { - return output, errors.Wrapf(err, "unable to list proxy endpoint in vpc %q", serverInfo.VpcId) - } - if len(lr.Data) > 0 { - // TODO Choose strictly - proxyEndpointId, _ = lr.Data[0].GetString("id") - } - } - if len(proxyEndpointId) == 0 { - return output, httperrors.NewInputParameterError("can't find suitable proxy endpoint for server %s, please connect with admin to create one", serverInfo.serverDetails.Name) - } - input.ProxyEndpointId = proxyEndpointId - } - ag, _ := getArgGenerator(MonitorAgent) - args, err := ag(ctx, input, serverInfo.serverDetails) + sa, err := ScriptApplyManager.createScriptApply(ctx, s.Id, input.ServerID, nil, argsGenerator) if err != nil { - return output, errors.Wrapf(err, "unable to get args of server %s", serverInfo.ServerId) - } - sa, err := ScriptApplyManager.createScriptApply(ctx, s.Id, serverInfo.ServerId, input.ProxyEndpointId, input.EipFirst, args) - if err != nil { - return output, errors.Wrapf(err, "unable to apply script to server %s", serverInfo.ServerId) + return output, errors.Wrapf(err, "unable to apply script to server %s", input.ServerID) } err = sa.StartApply(ctx, userCred) if err != nil { - return output, errors.Wrapf(err, "unable to apply script to server %s", serverInfo.ServerId) + return output, errors.Wrapf(err, "unable to apply script to server %s", input.ServerID) } output.ScriptApplyId = sa.Id return output, nil @@ -312,34 +158,3 @@ type sServerInfo struct { NetworkIds []string serverDetails *comapi.ServerDetails } - -func (s *SScript) checkServer(ctx context.Context, userCred mcclient.TokenCredential, serverId string) (sServerInfo, error) { - session := auth.GetSessionWithInternal(ctx, userCred, "", "") - // check server - data, err := modules.Servers.Get(session, serverId, nil) - if err != nil { - if httputils.ErrorCode(err) == 404 { - return sServerInfo{}, httperrors.NewInputParameterError("no such server %s", serverId) - } - return sServerInfo{}, fmt.Errorf("unable to get server %s: %s", serverId, httputils.ErrorMsg(err)) - } - info := sServerInfo{} - var serverDetails comapi.ServerDetails - err = data.Unmarshal(&serverDetails) - if err != nil { - return info, errors.Wrap(err, "unable to unmarshal serverDetails") - } - if serverDetails.Status != comapi.VM_RUNNING { - return info, httperrors.NewInputParameterError("can only apply scripts to %s server", comapi.VM_RUNNING) - } - info.serverDetails = &serverDetails - info.ServerId = serverDetails.Id - - networkIds := sets.NewString() - for _, nic := range serverDetails.Nics { - networkIds.Insert(nic.NetworkId) - info.VpcId = nic.VpcId - } - info.NetworkIds = networkIds.UnsortedList() - return info, nil -} diff --git a/pkg/devtool/models/script_apply.go b/pkg/devtool/models/script_apply.go index 8da8caa3fe..37f5870a26 100644 --- a/pkg/devtool/models/script_apply.go +++ b/pkg/devtool/models/script_apply.go @@ -21,7 +21,6 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/pkg/errors" - "yunion.io/x/pkg/tristate" "yunion.io/x/pkg/util/sets" api "yunion.io/x/onecloud/pkg/apis/devtool" @@ -34,11 +33,10 @@ type SScriptApply struct { db.SStatusStandaloneResourceBase ScriptId string `width:"36" nullable:"false" index:"true"` GuestId string `width:"36" nullable:"false" index:"true"` - EipFirst tristate.TriState // - Args jsonutils.JSONObject - ProxyEndpointId string `width:"36" nullable:"false"` - TryTimes int + Args jsonutils.JSONObject + TryTimes int + ArgsGenerator string `width:"36" nullable:"false"` } type SScriptApplyManager struct { @@ -61,13 +59,12 @@ func init() { ScriptApplyManager.SetVirtualObject(ScriptApplyManager) } -func (sam *SScriptApplyManager) createScriptApply(ctx context.Context, scriptId, guestId, proxyEndpointId string, eipFirst bool, args map[string]interface{}) (*SScriptApply, error) { +func (sam *SScriptApplyManager) createScriptApply(ctx context.Context, scriptId, guestId string, args map[string]interface{}, argsGenerator string) (*SScriptApply, error) { sa := &SScriptApply{ - ScriptId: scriptId, - GuestId: guestId, - EipFirst: tristate.NewFromBool(eipFirst), - ProxyEndpointId: proxyEndpointId, - Args: jsonutils.Marshal(args), + ScriptId: scriptId, + GuestId: guestId, + Args: jsonutils.Marshal(args), + ArgsGenerator: argsGenerator, } err := ScriptApplyManager.TableSpec().Insert(ctx, sa) sa.SetModelManager(ScriptApplyManager, sa) @@ -121,7 +118,7 @@ func (sa *SScriptApply) startApplyScriptTask(ctx context.Context, userCred mccli return nil } -func (sa *SScriptApply) StopApply(userCred mcclient.TokenCredential, record *SScriptApplyRecord, success bool, reason string) error { +func (sa *SScriptApply) StopApply(userCred mcclient.TokenCredential, record *SScriptApplyRecord, success bool, failCode string, reason string) error { var status string if success { status = api.SCRIPT_APPLY_STATUS_READY @@ -131,7 +128,7 @@ func (sa *SScriptApply) StopApply(userCred mcclient.TokenCredential, record *SSc } else { status = api.SCRIPT_APPLY_RECORD_FAILED if record != nil { - record.Fail(reason) + record.Fail(failCode, reason) } } sa.SetStatus(userCred, status, "") diff --git a/pkg/devtool/models/script_apply_record.go b/pkg/devtool/models/script_apply_record.go index 0482c25605..3c98ef3324 100644 --- a/pkg/devtool/models/script_apply_record.go +++ b/pkg/devtool/models/script_apply_record.go @@ -35,6 +35,7 @@ type SScriptApplyRecord struct { StartTime time.Time `list:"user"` EndTime time.Time `list:"user"` Reason string `list:"user"` + FailCode string `list:"user"` } type SScriptApplyRecordManager struct { @@ -159,20 +160,21 @@ func (sar *SScriptApplyRecord) GetOwnerId() mcclient.IIdentityProvider { return obj.GetOwnerId() } -func (sar *SScriptApplyRecord) SetResult(status, reason string) error { +func (sar *SScriptApplyRecord) SetResult(status, failCode, reason string) error { _, err := db.Update(sar, func() error { sar.Status = status sar.Reason = reason + sar.FailCode = failCode sar.EndTime = time.Now() return nil }) return err } -func (sar *SScriptApplyRecord) Fail(reason string) error { - return sar.SetResult(api.SCRIPT_APPLY_RECORD_FAILED, reason) +func (sar *SScriptApplyRecord) Fail(code string, reason string) error { + return sar.SetResult(api.SCRIPT_APPLY_RECORD_FAILED, code, reason) } func (sar *SScriptApplyRecord) Succeed(reason string) error { - return sar.SetResult(api.SCRIPT_APPLY_RECORD_SUCCEED, reason) + return sar.SetResult(api.SCRIPT_APPLY_RECORD_SUCCEED, "", reason) } diff --git a/pkg/devtool/tasks/apply_script_task.go b/pkg/devtool/tasks/apply_script_task.go index c5e7c0ddeb..c27d7eaab4 100644 --- a/pkg/devtool/tasks/apply_script_task.go +++ b/pkg/devtool/tasks/apply_script_task.go @@ -28,9 +28,11 @@ import ( ansible_api "yunion.io/x/onecloud/pkg/apis/ansible" cloudproxy_api "yunion.io/x/onecloud/pkg/apis/cloudproxy" comapi "yunion.io/x/onecloud/pkg/apis/compute" + devtool_api "yunion.io/x/onecloud/pkg/apis/devtool" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" "yunion.io/x/onecloud/pkg/devtool/models" + "yunion.io/x/onecloud/pkg/devtool/utils" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/modules" @@ -45,23 +47,36 @@ func init() { taskman.RegisterTask(ApplyScriptTask{}) } +var ErrServerNotSshable = errors.Error("server is not sshable") + func (self *ApplyScriptTask) taskFailed(ctx context.Context, sa *models.SScriptApply, sar *models.SScriptApplyRecord, err error) { - err = sa.StopApply(self.UserCred, sar, false, err.Error()) + var failCode string + switch errors.Cause(err) { + case ErrServerNotSshable: + failCode = devtool_api.SCRIPT_APPLY_RECORD_FAILCODE_SSHABLE + case utils.ErrCannotReachInfluxbd: + failCode = devtool_api.SCRIPT_APPLY_RECORD_FAILCODE_INFLUXDB + default: + failCode = devtool_api.SCRIPT_APPLY_RECORD_FAILCODE_OTHERS + } + err = sa.StopApply(self.UserCred, sar, false, failCode, err.Error()) if err != nil { log.Errorf("unable to StopApply script %s to server %s", sa.ScriptId, sa.GuestId) self.SetStageFailed(ctx, jsonutils.NewString(err.Error())) return } - // restart - err = sa.StartApply(ctx, self.UserCred) - if err != nil { - log.Errorf("unable to StartApply script %s to server %s", sa.ScriptId, sa.GuestId) + if failCode == devtool_api.SCRIPT_APPLY_RECORD_FAILCODE_OTHERS { + // restart + err = sa.StartApply(ctx, self.UserCred) + if err != nil { + log.Errorf("unable to StartApply script %s to server %s", sa.ScriptId, sa.GuestId) + } } self.SetStageFailed(ctx, jsonutils.NewString(err.Error())) } func (self *ApplyScriptTask) taskSuccess(ctx context.Context, sa *models.SScriptApply, sar *models.SScriptApplyRecord) { - err := sa.StopApply(self.UserCred, sar, true, "") + err := sa.StopApply(self.UserCred, sar, true, "", "") if err != nil { log.Errorf("unable to StopApply script %s to server %s", sa.ScriptId, sa.GuestId) self.SetStageComplete(ctx, nil) @@ -103,7 +118,11 @@ func (self *ApplyScriptTask) OnInit(ctx context.Context, obj db.IStandaloneModel return } if !sshable.ok { - self.taskFailed(ctx, sa, sar, fmt.Errorf("server %s is not sshable: %s", serverDetail.Id, sshable.reason)) + var err error = ErrServerNotSshable + if len(sshable.reason) > 0 { + err = errors.Wrap(err, sshable.reason) + } + self.taskFailed(ctx, sa, sar, err) return } // make sure user @@ -116,49 +135,79 @@ func (self *ApplyScriptTask) OnInit(ctx context.Context, obj db.IStandaloneModel default: user = "cloudroot" } - // create local forward - createP := jsonutils.NewDict() - createP.Set("type", jsonutils.NewString(cloudproxy_api.FORWARD_TYPE_LOCAL)) - createP.Set("remote_port", jsonutils.NewInt(22)) - createP.Set("server_id", jsonutils.NewString(serverDetail.Id)) - forward, err := cloudproxy.Forwards.PerformClassAction(session, "create-from-server", createP) - if err != nil { - self.taskFailed(ctx, sa, sar, errors.Wrapf(err, "fail to create local forward from server %q", serverDetail.Id)) - return + var host ansible_api.AnsibleHost + var forwardId string + if len(sshable.proxyEndpointId) == 0 { + host = ansible_api.AnsibleHost{ + User: user, + IP: sshable.host, + Port: sshable.port, + Name: serverDetail.Name, + } + } else { + // create local forward + createP := jsonutils.NewDict() + createP.Set("type", jsonutils.NewString(cloudproxy_api.FORWARD_TYPE_LOCAL)) + createP.Set("remote_port", jsonutils.NewInt(22)) + createP.Set("server_id", jsonutils.NewString(serverDetail.Id)) + + forward, err := cloudproxy.Forwards.PerformClassAction(session, "create-from-server", createP) + if err != nil { + self.taskFailed(ctx, sa, sar, errors.Wrapf(err, "fail to create local forward from server %q", serverDetail.Id)) + return + } + + port, _ := forward.Int("bind_port") + forwardId, _ = forward.GetString("id") + agentId, _ := forward.GetString("proxy_agent_id") + agent, err := cloudproxy.ProxyAgents.Get(session, agentId, nil) + if err != nil { + self.clearLocalForward(session, forwardId) + self.taskFailed(ctx, sa, sar, errors.Wrapf(err, "fail to get proxy agent %q", agentId)) + return + } + address, _ := agent.GetString("advertise_addr") + // check proxy forward + if ok := self.ensureLocalForwardWork(address, int(port)); !ok { + self.clearLocalForward(session, forwardId) + self.taskFailed(ctx, sa, sar, errors.Error("The created local forward is actually not usable")) + return + } + host = ansible_api.AnsibleHost{ + User: user, + IP: address, + Port: int(port), + Name: serverDetail.Name, + } } - port, _ := forward.Int("bind_port") - forwardId, _ := forward.GetString("id") - agentId, _ := forward.GetString("proxy_agent_id") - agent, err := cloudproxy.ProxyAgents.Get(session, agentId, nil) - if err != nil { - self.clearLocalForward(session, forwardId) - self.taskFailed(ctx, sa, sar, errors.Wrapf(err, "fail to get proxy agent %q", agentId)) - return - } - address, _ := agent.GetString("advertise_addr") - host := ansible_api.AnsibleHost{ - User: user, - IP: address, - Port: int(port), - Name: serverDetail.Name, - } + // genrate args params = jsonutils.NewDict() - params.Set("args", sa.Args) + if len(sa.ArgsGenerator) == 0 { + params.Set("args", sa.Args) + } else { + generator, ok := utils.GetArgGenerator(sa.ArgsGenerator) + if !ok { + params.Set("args", sa.Args) + } + arg, err := generator(ctx, sa.GuestId, sshable.proxyEndpointId, &host) + if err != nil { + self.clearLocalForward(session, forwardId) + self.taskFailed(ctx, sa, sar, err) + return + } + params.Set("args", jsonutils.Marshal(arg)) + } + params.Set("host", jsonutils.Marshal(host)) + // fetch ansible playbook reference id updateData := jsonutils.NewDict() updateData.Set("script_apply_record_id", jsonutils.NewString(sar.GetId())) updateData.Set("proxy_forward_id", jsonutils.NewString(forwardId)) - - // check proxy forward - if ok := self.ensureLocalForwardWork(address, int(port)); !ok { - self.clearLocalForward(session, forwardId) - self.taskFailed(ctx, sa, sar, errors.Error("The created local forward is actually not usable")) - return - } self.SetStage("OnAnsiblePlaybookComplete", updateData) + // Inject Task Header session.Header = self.GetTaskRequestHeader() _, err = modules.AnsiblePlaybookReference.PerformAction(session, s.PlaybookReferenceId, "run", params) @@ -170,38 +219,59 @@ func (self *ApplyScriptTask) OnInit(ctx context.Context, obj db.IStandaloneModel } type sSSHable struct { - user string ok bool reason string + + user string + + proxyEndpointId string + proxyAgentId string + + host string + port int } +// func (self *ApplyScriptTask) ansibleHost(session modules.SS) + func (self *ApplyScriptTask) checkSshable(session *mcclient.ClientSession, serverId string) (sSSHable, error) { data, err := modules.Servers.GetSpecific(session, serverId, "sshable", nil) if err != nil { return sSSHable{}, errors.Wrapf(err, "unable to get sshable info of server %s", serverId) } - log.Debugf("data to chech sshable:\n %s", data) - methodTrieds, _ := data.GetArray("method_tried") - sshable := sSSHable{} - reasons := make([]string, 0, len(methodTrieds)) - for _, methodTried := range methodTrieds { - ok, _ := methodTried.Bool("sshable") - if ok { - sshable.ok = true - break + var sshableOutput comapi.GuestSshableOutput + err = data.Unmarshal(&sshableOutput) + if err != nil { + return sSSHable{}, errors.Wrapf(err, "unable to marshal output of server sshable: %s", data) + } + sshable := sSSHable{ + user: sshableOutput.User, + } + reasons := make([]string, 0, len(sshableOutput.MethodTried)) + for _, methodTried := range sshableOutput.MethodTried { + if !methodTried.Sshable { + reasons = append(reasons, methodTried.Reason) + continue + } + sshable.ok = true + switch methodTried.Method { + case comapi.MethodDirect, comapi.MethodEIP, comapi.MethodDNAT: + sshable.host = methodTried.Host + sshable.port = methodTried.Port + case comapi.MethodProxyForward: + sshable.proxyAgentId = methodTried.ForwardDetails.ProxyAgentId + sshable.proxyEndpointId = methodTried.ForwardDetails.ProxyEndpointId } - reason, _ := methodTried.GetString("reason") - reasons = append(reasons, reason) } if !sshable.ok { sshable.reason = strings.Join(reasons, "; ") - } else { - sshable.user, _ = data.GetString("user") } return sshable, nil } func (self *ApplyScriptTask) clearLocalForward(s *mcclient.ClientSession, forwardId string) { + if len(forwardId) == 0 { + return + } _, err := cloudproxy.Forwards.Delete(s, forwardId, nil) if err != nil { log.Errorf("unable to delete proxy forward %s", forwardId) diff --git a/pkg/devtool/utils/arg_generator.go b/pkg/devtool/utils/arg_generator.go new file mode 100644 index 0000000000..e1f2842aba --- /dev/null +++ b/pkg/devtool/utils/arg_generator.go @@ -0,0 +1,36 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "context" + "sync" +) + +type argGenerator func(ctx context.Context, serverId, proxyEndpointId string, others interface{}) (map[string]interface{}, error) + +var argGenerators = &sync.Map{} + +func RegisterArgGenerator(name string, ag argGenerator) { + argGenerators.Store(name, ag) +} + +func GetArgGenerator(name string) (argGenerator, bool) { + v, ok := argGenerators.Load(name) + if !ok { + return nil, ok + } + return v.(argGenerator), ok +} diff --git a/pkg/devtool/utils/doc.go b/pkg/devtool/utils/doc.go new file mode 100644 index 0000000000..f0e9e6a78e --- /dev/null +++ b/pkg/devtool/utils/doc.go @@ -0,0 +1,15 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils // import "yunion.io/x/onecloud/pkg/devtool/utils" diff --git a/pkg/devtool/utils/influxdb_url.go b/pkg/devtool/utils/influxdb_url.go new file mode 100644 index 0000000000..c4826bc1cd --- /dev/null +++ b/pkg/devtool/utils/influxdb_url.go @@ -0,0 +1,353 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "context" + "fmt" + "net/url" + "time" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/util/sets" + + ansible_api "yunion.io/x/onecloud/pkg/apis/ansible" + proxy_api "yunion.io/x/onecloud/pkg/apis/cloudproxy" + comapi "yunion.io/x/onecloud/pkg/apis/compute" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/mcclient/modules/cloudproxy" + "yunion.io/x/onecloud/pkg/util/ansible" + "yunion.io/x/onecloud/pkg/util/httputils" +) + +type sServerInfo struct { + ServerId string + VpcId string + NetworkIds []string + serverDetails *comapi.ServerDetails +} + +type sProxyEndpoint struct { + Id string + Address string +} + +func proxyEndpoints(ctx context.Context, proxyEndpointId string, info sServerInfo) ([]sProxyEndpoint, error) { + pes := make([]sProxyEndpoint, 0) + session := auth.GetAdminSession(ctx, "", "") + if len(proxyEndpointId) > 0 { + ep, err := cloudproxy.ProxyEndpoints.Get(session, proxyEndpointId, nil) + if err != nil { + return nil, errors.Wrapf(err, "unable to get proxy endpoint %s", proxyEndpointId) + } + address, _ := ep.GetString("intranet_ip_addr") + pes = append(pes, sProxyEndpoint{proxyEndpointId, address}) + return pes, nil + } + proxyEndpointIds := sets.NewString() + for _, netId := range info.NetworkIds { + filter := jsonutils.NewDict() + filter.Set("network_id", jsonutils.NewString(netId)) + lr, err := cloudproxy.ProxyEndpoints.List(session, filter) + if err != nil { + return nil, errors.Wrapf(err, "unable to list proxy endpoint in network %q", netId) + } + for i := range lr.Data { + proxyEndpointId, _ := lr.Data[i].GetString("id") + address, _ := lr.Data[i].GetString("intranet_ip_addr") + if proxyEndpointIds.Has(proxyEndpointId) { + continue + } + pes = append(pes, sProxyEndpoint{proxyEndpointId, address}) + proxyEndpointIds.Insert(proxyEndpointId) + } + } + filter := jsonutils.NewDict() + filter.Set("vpc_id", jsonutils.NewString(info.VpcId)) + lr, err := cloudproxy.ProxyEndpoints.List(session, filter) + if err != nil { + return nil, errors.Wrapf(err, "unable to list proxy endpoint in vpc %q", info.VpcId) + } + for i := range lr.Data { + proxyEndpointId, _ := lr.Data[i].GetString("id") + address, _ := lr.Data[i].GetString("intranet_ip_addr") + if proxyEndpointIds.Has(proxyEndpointId) { + continue + } + pes = append(pes, sProxyEndpoint{proxyEndpointId, address}) + proxyEndpointIds.Insert(proxyEndpointId) + } + return pes, nil +} + +func getServerInfo(ctx context.Context, serverId string) (sServerInfo, error) { + // check server + session := auth.GetAdminSession(ctx, "", "") + data, err := modules.Servers.Get(session, serverId, nil) + if err != nil { + if httputils.ErrorCode(err) == 404 { + return sServerInfo{}, httperrors.NewInputParameterError("no such server %s", serverId) + } + return sServerInfo{}, fmt.Errorf("unable to get server %s: %s", serverId, httputils.ErrorMsg(err)) + } + info := sServerInfo{} + var serverDetails comapi.ServerDetails + err = data.Unmarshal(&serverDetails) + if err != nil { + return info, errors.Wrap(err, "unable to unmarshal serverDetails") + } + if serverDetails.Status != comapi.VM_RUNNING { + return info, httperrors.NewInputParameterError("can only apply scripts to %s server", comapi.VM_RUNNING) + } + info.serverDetails = &serverDetails + info.ServerId = serverDetails.Id + + networkIds := sets.NewString() + for _, nic := range serverDetails.Nics { + networkIds.Insert(nic.NetworkId) + info.VpcId = nic.VpcId + } + info.NetworkIds = networkIds.UnsortedList() + return info, nil +} + +func convertInfluxdbUrl(ctx context.Context, pUrl string, endpointId string) (port int64, recycle func() error, err error) { + session := auth.AdminSessionWithInternal(ctx, "", "", "") + filter := jsonutils.NewDict() + filter.Set("proxy_endpoint_id", jsonutils.NewString(endpointId)) + filter.Set("opaque", jsonutils.NewString(pUrl)) + filter.Set("scope", jsonutils.NewString("system")) + lr, err := cloudproxy.Forwards.List(session, filter) + if err != nil { + return 0, nil, errors.Wrap(err, "failed to list forward") + } + if len(lr.Data) > 0 { + port, _ = lr.Data[0].Int("bind_port") + } else { + var rUrl *url.URL + rUrl, err = url.Parse(pUrl) + if err != nil { + err = errors.Wrap(err, "invalid influxdbUrl?") + return + } + // create one + createP := jsonutils.NewDict() + createP.Set("proxy_endpoint", jsonutils.NewString(endpointId)) + createP.Set("type", jsonutils.NewString(proxy_api.FORWARD_TYPE_REMOTE)) + createP.Set("remote_addr", jsonutils.NewString(rUrl.Hostname())) + createP.Set("remote_port", jsonutils.NewString(rUrl.Port())) + createP.Set("generate_name", jsonutils.NewString("influxdb proxy")) + createP.Set("opaque", jsonutils.NewString(pUrl)) + var forward jsonutils.JSONObject + forward, err = cloudproxy.Forwards.Create(session, createP) + if err != nil { + err = errors.Wrapf(err, "unable to create forward with create params %s", createP.String()) + return + } + forwardId, _ := forward.GetString("id") + recycle = func() error { + _, err := cloudproxy.Forwards.Delete(session, forwardId, nil) + return err + } + port, _ = forward.Int("bind_port") + } + return +} + +func checkProxyEndpoint(ctx context.Context, influxdbUrl, proxyEndpointId, address string, host *ansible_api.AnsibleHost) (string, error) { + port, recycle, err := convertInfluxdbUrl(ctx, influxdbUrl, proxyEndpointId) + if err != nil { + return "", err + } + nUrl := fmt.Sprintf("https://%s:%d", address, port) + ok, err := checkUrl(ctx, nUrl, host) + if err != nil { + return "", errors.Wrapf(err, "check url %q", nUrl) + } + if !ok { + if recycle != nil { + err := recycle() + if err != nil { + return "", errors.Wrapf(err, "unble to recycle remote forward of proxyEndpoint %s", proxyEndpointId) + } + } + return "", nil + } + return nUrl, nil +} + +func findValidInfluxdbUrl(ctx context.Context, influxdbUrl, proxyEndpointId string, info sServerInfo, host *ansible_api.AnsibleHost) (string, error) { + if len(proxyEndpointId) > 0 { + pes, err := proxyEndpoints(ctx, proxyEndpointId, info) + if err != nil { + return "", err + } + url, err := checkProxyEndpoint(ctx, influxdbUrl, proxyEndpointId, pes[0].Address, host) + if err != nil { + return "", err + } + if len(url) > 0 { + return url, nil + } + } + pes, err := proxyEndpoints(ctx, "", info) + if err != nil { + return "", err + } + for _, pe := range pes { + url, err := checkProxyEndpoint(ctx, influxdbUrl, pe.Id, pe.Address, host) + if err != nil { + return "", err + } + if len(url) > 0 { + return url, nil + } + } + // check direct + ok, err := checkUrl(ctx, influxdbUrl, host) + if err != nil { + return "", err + } + if ok { + return influxdbUrl, nil + } + return "", nil +} + +func checkUrl(ctx context.Context, url string, host *ansible_api.AnsibleHost) (bool, error) { + session := auth.GetAdminSession(ctx, "", "") + ahost := ansible.Host{} + ahost.Name = host.IP + ahost.Vars = map[string]string{ + "ansible_port": fmt.Sprintf("%d", host.Port), + "ansible_user": host.User, + } + mod := ansible.Module{ + Name: "uri", + Args: []string{ + fmt.Sprintf("url=%s/ping", url), + "method=GET", + "status_code=204", + "validate_certs=no", + }, + } + + playbook := ansible.NewPlaybook() + playbook.Inventory = ansible.Inventory{ + Hosts: []ansible.Host{ + ahost, + }, + } + playbook.Modules = []ansible.Module{ + mod, + } + apCreateInput := ansible_api.AnsiblePlaybookCreateInput{ + Name: db.DefaultUUIDGenerator(), + Playbook: *playbook, + } + apb, err := modules.AnsiblePlaybooks.Create(session, apCreateInput.JSON(apCreateInput)) + if err != nil { + return false, errors.Wrap(err, "create ansible playbook") + } + id, _ := apb.GetString("id") + defer func() { + _, err := modules.AnsiblePlaybooks.Delete(session, id, nil) + if err != nil { + log.Errorf("unable to delete ansibleplaybook %s: %v", id, err) + } + }() + times, waitTimes := 0, time.Second + for times < 5 { + time.Sleep(waitTimes) + times++ + waitTimes += time.Second * time.Duration(times) + apd, err := modules.AnsiblePlaybooks.GetSpecific(session, id, "status", nil) + if err != nil { + return false, errors.Wrapf(err, "unable to get ansibleplaybook %s status", id) + } + status, _ := apd.GetString("status") + switch status { + case ansible_api.AnsiblePlaybookStatusInit, ansible_api.AnsiblePlaybookStatusRunning: + continue + case ansible_api.AnsiblePlaybookStatusFailed, ansible_api.AnsiblePlaybookStatusCanceled, ansible_api.AnsiblePlaybookStatusUnknown: + return false, nil + case ansible_api.AnsiblePlaybookStatusSucceeded: + return true, nil + } + } + return false, nil +} + +var influxdbUrl string + +func getInfluxdbUrl(ctx context.Context) (string, error) { + if len(influxdbUrl) > 0 { + return influxdbUrl, nil + } + session := auth.GetAdminSession(ctx, "", "") + params := jsonutils.NewDict() + params.Set("interface", jsonutils.NewString("public")) + params.Set("service", jsonutils.NewString("influxdb")) + ret, err := modules.EndpointsV3.List(session, params) + if err != nil { + return "", err + } + if len(ret.Data) == 0 { + return "", fmt.Errorf("no sucn endpoint with 'internal' interface and 'influxdb' service") + } + url, _ := ret.Data[0].GetString("url") + return url, nil +} + +var ErrCannotReachInfluxbd = errors.Error("no suitable network to reach influxdb") + +func GetArgs(ctx context.Context, serverId, proxyEndpointId string, others interface{}) (map[string]interface{}, error) { + host, ok := others.(*ansible_api.AnsibleHost) + if !ok { + return nil, errors.Error("unknown others, want *AnsibleHost") + } + info, err := getServerInfo(ctx, serverId) + if err != nil { + return nil, errors.Wrapf(err, "unable to get serverInfo of server %s", serverId) + } + + influxdbUrl, err := getInfluxdbUrl(ctx) + if err != nil { + return nil, errors.Wrap(err, "unable to get influxdbUrl") + } + influxdbUrl, err = findValidInfluxdbUrl(ctx, influxdbUrl, proxyEndpointId, info, host) + if err != nil { + return nil, errors.Wrapf(err, "unable to convertInfluxdbUrl %s", influxdbUrl) + } + if len(influxdbUrl) == 0 { + return nil, errors.Wrap(ErrCannotReachInfluxbd, "please create usable Proxy Endpoint for server and try again") + } + vmId := info.serverDetails.Id + tenantId := info.serverDetails.ProjectId + domainId := info.serverDetails.DomainId + ret := map[string]interface{}{ + "influxdb_url": influxdbUrl, + "influxdb_name": "telegraf", + "onecloud_vm_id": vmId, + "onecloud_tenant_id": tenantId, + "onecloud_domain_id": domainId, + } + return ret, nil +}