gomaxcompute/instance.go at develop · sql-machine-learning/gomaxcompute · GitHub

History

143 lines (125 loc) · 3.49 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

package gomaxcompute

import (

"encoding/base64"

"encoding/xml"

"fmt"

"strings"

"github.com/pkg/errors"

)

type instanceStatus struct {

XMLName xml.Name `xml:"Instance"`

Status string `xml:"Status"`

}

type Result struct {

Content string `xml:",cdata"`

Transform string `xml:"Transform,attr"`

Format string `xml:"Format,attr"`

}

type instanceResult struct {

XMLName xml.Name `xml:"Instance"`

Result Result `xml:"Tasks>Task>Result"`

}

type instanceErrorMessage struct {

CDATA string `xml:",cdata"`

}

type instanceError struct {

XMLName xml.Name `xml:"Error"`

Code string `xml:"Code"`

Message instanceErrorMessage `xml:"Message"`

RequestId string `xml:"RequestId"`

HostId string `xml:"HostId"`

}

// instance types：SQL

func (conn *odpsConn) createInstance(job *odpsJob) (string, error) {

if job == nil {

return "", errors.New("nil job")

}

// Create

res, err := conn.request(methodPost, conn.resource("/instances"), job.XML())

if err != nil {

return "", err

}

if _, err = parseResponse(res); err != nil && err != errNilBody {

return "", err

}

// Parse response header "Location" to get instance ID

ins := location2InstanceID(res.Header.Get("Location"))

if ins == "" {

return "", errors.New("no instance id")

}

return ins, nil

}

// parse instance id

func location2InstanceID(location string) string {

pieces := strings.Split(location, "/")

if len(pieces) < 2 {

return ""

}

return pieces[len(pieces)-1]

}

// instance status：Running/Suspended/Terminated

func (conn *odpsConn) getInstanceStatus(instanceID string) (string, error) {

res, err := conn.request(methodGet, conn.resource("/instances/"+instanceID), nil)

if err != nil {

return "", err

}

body, err := parseResponse(res)

if err != nil {

return "", err

}

var is instanceStatus

err = xml.Unmarshal(body, &is)

if err != nil {

return "", err

}

return is.Status, nil

}

// getInstanceResult is valid while instance status is `Terminated`

// notice: records up to 10000 by limitation, and result type is string

func (conn *odpsConn) getInstanceResult(instanceID string) (string, error) {

rsc := conn.resource("/instances/"+instanceID, pair{k: "result"})

rsp, err := conn.request(methodGet, rsc, nil)

if err != nil {

return "", err

}

body, err := parseResponse(rsp)

if err != nil {

return "", err

}

return decodeInstanceResult(body)

}

func decodeInstanceResult(result []byte) (string, error) {

var ir instanceResult

if err := xml.Unmarshal(result, &ir); err != nil {

return "", err

}

if ir.Result.Format == "text" {

log.Debug(ir.Result.Content)

// ODPS errors are text begin with "ODPS-"

if strings.HasPrefix(ir.Result.Content, "ODPS-") {

code, err := parseErrorCode(ir.Result.Content)

if err != nil {

return "", errors.WithStack(errors.New(ir.Result.Content))

}

return "", &MaxcomputeError{code, ir.Result.Content}

}

// FIXME(tony): the result non-query statement usually in text format.

// Go's database/sql API only supports lastId and affectedRows.

return "", nil

}

if ir.Result.Format != "csv" {

return "", errors.WithStack(fmt.Errorf("unsupported format %v", ir.Result.Format))

}

switch ir.Result.Transform {

case "":

return ir.Result.Content, nil

case "Base64":

content, err := base64.StdEncoding.DecodeString(ir.Result.Content)

if err != nil {

return "", err

}

return string(content), err

default:

return "", errors.WithStack(fmt.Errorf("unsupported transform %v", ir.Result.Transform))

}

gomaxcompute/instance.go at develop · sql-machine-learning/gomaxcompute · GitHub

Uh oh!

Expand file tree

Breadcrumbs

instance.go

Latest commit

History

Breadcrumbs

instance.go

File metadata and controls