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 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/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/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, 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 f609425030..27ab9c3b2a 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 +}