go - 为什么我会从此 API 调用中收到意外的 EOF？

Question

我有以下作为 AWS Lambda cron 运行的 Go 代码，但我不确定为什么会出现此错误:

sls logs --stage prod --region eu-west-1 --function esCronFn
2018/12/12 12:07:01 unexpected EOF
2018/12/12 12:07:01 unexpected EOF
END RequestId: 6bf33d28-fe03-11e8-949d-f39174c57cab
REPORT RequestId: 6bf33d28-fe03-11e8-949d-f39174c57cab  Duration: 464734.47 ms  Billed Duration: 464800 ms      Memory Size: 256 MB     Max Memory Used: 257 MB

RequestId: 6bf33d28-fe03-11e8-949d-f39174c57cab Process exited before completing request

这是我的 main.go - 它基本上连接到外部 API 并提取我正在处理并上传到 S3 存储桶的记录。

主要包

import (
    "bytes"
    "encoding/csv"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "log"
    "net/http"
    "net/url"
    "os"
    "strings"
    "time"

    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
    "github.com/aws/aws-sdk-go/service/s3/s3iface"
)

var (
    // ENDPOINT is the endpoint from which incomplete CSV records are downloaded
    ENDPOINT = os.Getenv("ENDPOINT")

    PARSED_ENDPOINT *url.URL

    // TOKEN authenticates requests sent to eshot API
    TOKEN = os.Getenv("TOKEN")

    // BUCKET is the S3 bucket to which CSV files are uploaded
    BUCKET = os.Getenv("BUCKET")

    svc s3iface.S3API
)

// Record is the JSON response returned by a successful request to API
type EsRecord struct {
    Salutation   string    `json:"Salutation"`
    Firstname    string    `json:"Firstname"`
    Lastname     string    `json:"Lastname"`
    Company      string    `json:"Company"`
    EEA          string    `json:"EEA"`
    ModifiedDate time.Time `json:"ModifiedDate"`
    SubaccountID string    `json:"SubaccountId"`
    Email        string    `json:"Email"`
}

// CsvData holds reference to underlying buffer and the csv writer
type CsvData struct {
    Buffer *bytes.Buffer
    Writer *csv.Writer
}

func init() {
    today := time.Now()

    // If ENDPOINT is empty, It'll use this hardcoded endpoint. The ENDPOINT variable should not contain any text after "ModifiedDate gt". The actual date is currentDay-1
    if ENDPOINT == "" {
        ENDPOINT = "https://rest-api.domain.tld/Export/?$select=Email,Firstname,Lastname,SubaccountId,EEA,ModifiedDate&$filter=(EEA eq '' or EEA eq null) and ModifiedDate gt"
    }

    // Append CurrentDay-1 in YYYY-MM-DDTHH:MM:SSZ format.
    // The time is NOT in UTC. It's the local time of the machine on which lambda function was running
    ENDPOINT = fmt.Sprintf("%s %sT00:00:00Z", ENDPOINT, today.AddDate(0, 0, -1).Format("2006-01-02"))

    var err error
    PARSED_ENDPOINT, err = url.Parse(ENDPOINT)
    if err != nil {
        log.Fatalln("Invalid $ENDPOINT", err)
    }

    PARSED_ENDPOINT.RawQuery = QueryEscape(PARSED_ENDPOINT.RawQuery)
}

func main() {
    if TOKEN == "" {
        log.Fatalln("$TOKEN is empty")
    }
    if BUCKET == "" {
        log.Fatalln("$BUCKET is empty")
    }
    // Create S3 session
    svc = s3iface.S3API(s3.New(session.Must(session.NewSession())))

    lambda.Start(CreateAndUploadCsvToS3)
}

func CreateAndUploadCsvToS3() error {

    resp, err := fetchRecords()
    if err != nil {
        return fmt.Errorf("error in fetching records: %s", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        b, _ := ioutil.ReadAll(resp.Body)
        return fmt.Errorf("api returned non 200 response(%d), URL: %s, %s", resp.StatusCode, PARSED_ENDPOINT.String(), string(b))
    }

    // API returns array of EshotRecord
    esRecords := []EsRecord{}

    err = json.NewDecoder(resp.Body).Decode(&esRecords)
    if err != nil {
        b, _ := ioutil.ReadAll(resp.Body)
        return fmt.Errorf("error in parsing response %s: %s", err, string(b))
    }

    recordsMap := ParseEsRecordsJSON(esRecords)

    ct := time.Now().String()
    for k, v := range recordsMap {

        key := fmt.Sprintf("%s_%s.csv", k, ct)

        _, err := svc.PutObject(&s3.PutObjectInput{
            Bucket: aws.String(BUCKET),
            // Key is in format, <subaccountid>_<current timestamp>.csv
            Key:  aws.String(key),
            Body: bytes.NewReader(v.Buffer.Bytes()),
        })
        if err != nil {
            return fmt.Errorf("error in uploading %s: %s", key, err)
        }
    }

    return nil
}

// ParseEsRecordsJSON takes an array of EsRecord
// Seperates each record by subAccountId
// Creates CSV files for each SubAccountId
// Returns the hashmap
func ParseEsRecordsJSON(esRecords []EsRecord) map[string]CsvData {
    recordsMap := make(map[string]CsvData)

    for _, v := range esRecords {
        // If v.SubaccountID was encountered for the first time
        // 1. Create a Buffer
        // 2. Write CSV headers to this buffer
        // 3. Store reference to this buffer and csv writer in hashmap
        if _, ok := recordsMap[v.SubaccountID]; !ok {
            var buf bytes.Buffer

            writer := csv.NewWriter(&buf)
            // Write CSV headers
            err := writer.Write([]string{"Firstname", "Lastname", "Email"})
            if err != nil {
                log.Printf("error occurred in inserting headers for subAccountId(%s): %s\n", v.SubaccountID, err)
            }

            // store reference to writer object for this subaccountid in hashmap
            recordsMap[v.SubaccountID] = CsvData{
                Buffer: &buf,
                Writer: writer,
            }
        }
        csvRef := recordsMap[v.SubaccountID]

        err := csvRef.Writer.Write([]string{v.Firstname, v.Lastname, v.Email})
        if err != nil {
            log.Printf("error occurred in inserting headers for subAccountId(%s): %s\n", v.SubaccountID, err)
        }
        csvRef.Writer.Flush()
    }
    return recordsMap
}

// FetchRecords makes a request to API and returns http.Response
func fetchRecords() (*http.Response, error) {
    req, err := http.NewRequest("GET", PARSED_ENDPOINT.String(), nil)
    if err != nil {
        return nil, err
    }

    req.Header.Set("Authorization", fmt.Sprintf("Token %s", TOKEN))
    client := &http.Client{}
    return client.Do(req)
}

// QueryEscape replaces URL unsafe characters as listed in HTTP RFC with their HEX values.
// The QueryEscape function in Go strictly adheres to the RFC and replaces all the characters listed in RFC with their HEX values.
// Curl/Postman only encodes parameters on a strict "need" only bases. Presumably, the endpoint does not seems to be working with Go's encoded string.
// This code escapes all the charactes and then performs uses string replace to make the URL more like what CURL would have done.
func QueryEscape(s string) string {
    s = url.QueryEscape(s)

    s = strings.Replace(s, "%2C", ",", -1)
    s = strings.Replace(s, "%24", "$", -1)
    s = strings.Replace(s, "%3D", "=", -1)
    s = strings.Replace(s, "+", "%20", -1)
    s = strings.Replace(s, "%26", "&", -1)
    s = strings.Replace(s, "%3A", ":", -1)

    return s
}

如果我将 ENDPOINT 更改为:

ENDPOINT = "https://rest-api.domain.tld/Export/?$select=Email,Firstname,Lastname,SubaccountId,EEA,ModifiedDate&$filter=(EEA eq '' or EEA eq null) and ModifiedDate gt"

到

ENDPOINT = "https://rest-api.domain.tld/Export/?$select=Email,Firstname,Lastname,SubaccountId,EEA,ModifiedDate&$filter=EEA eq '' and ModifiedDate gt"

我没有得到 EOF 错误，但是我没有得到完整的列表，运行 curl，我得到了我需要的数据，所以我不确定为什么我的代码失败了，如何最好地跟踪失败的地方？

Answer 1

问题是我的 lambda 函数只有 128Mb 内存，端点提供 130Mb 文件，所以增加这个值解决了这个问题