gorqlite/api.go

package gorqlite

/*
	this file has low level stuff:

	rqliteApiGet()
	rqliteApiPost()

	There is some code duplication between those and they should
	probably be combined into one function.

	nothing public here.

*/

import "bytes"
import "encoding/json"
import "errors"
import "fmt"
import "io/ioutil"
import "net/http"
import "time"

/* *****************************************************************

   method: rqliteApiGet() - for api_STATUS

	- lowest level interface - does not do any JSON unmarshaling
	- handles retries
	- handles timeouts

 * *****************************************************************/

func (conn *Connection) rqliteApiGet(apiOp apiOperation) ([]byte, error) {
	var responseBody []byte
	trace("%s: rqliteApiGet() called", conn.ID)

	// only api_STATUS now - maybe someday BACKUP
	if apiOp != api_STATUS {
		return responseBody, errors.New("rqliteApiGet() called for invalid api operation")
	}

	// just to be safe, check this
	peersToTry := conn.cluster.makePeerList()
	if len(peersToTry) < 1 {
		return responseBody, errors.New("I don't have any cluster info")
	}
	trace("%s: I have a peer list %d peers long", conn.ID, len(peersToTry))

	// failure log is used so that if all peers fail, we can say something
	// about why each failed
	failureLog := make([]string, 0)

PeerLoop:
	for peerNum, peerToTry := range peersToTry {
		trace("%s: attemping to contact peer %d", conn.ID, peerNum)
		// docs say default GET policy is up to 10 follows automatically
		url := conn.assembleURL(api_STATUS, peerToTry)
		req, err := http.NewRequest("GET", url, nil)
		if err != nil {
			trace("%s: got error '%s' doing http.NewRequest", conn.ID, err.Error())
			failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
			continue PeerLoop
		}
		trace("%s: http.NewRequest() OK")
		req.Header.Set("Content-Type", "application/json")
		client := &http.Client{}
		client.Timeout = time.Duration(conn.timeout) * time.Second
		response, err := client.Do(req)
		if err != nil {
			trace("%s: got error '%s' doing client.Do", conn.ID, err.Error())
			failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
			continue PeerLoop
		}
		defer response.Body.Close()
		trace("%s: client.Do() OK")
		responseBody, err := ioutil.ReadAll(response.Body)
		if err != nil {
			trace("%s: got error '%s' doing ioutil.ReadAll", conn.ID, err.Error())
			failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
			continue PeerLoop
		}
		trace("%s: ioutil.ReadAll() OK")
		if response.Status != "200 OK" {
			trace("%s: got code %s", conn.ID, response.Status)
			failureLog = append(failureLog, fmt.Sprintf("%s failed, got: %s", url, response.Status))
			continue PeerLoop
		}
		// if we got here, we succeeded
		trace("%s: api call OK, returning", conn.ID)
		return responseBody, nil
	}

	// if we got here, all peers failed.  Let's build a verbose error message
	var stringBuffer bytes.Buffer
	stringBuffer.WriteString("tried all peers unsuccessfully. here are the results:\n")
	for n, v := range failureLog {
		stringBuffer.WriteString(fmt.Sprintf("   peer #%d: %s\n", n, v))
	}
	return responseBody, errors.New(stringBuffer.String())
}

/* *****************************************************************

   method: rqliteApiPost() - for api_QUERY and api_WRITE

	- lowest level interface - does not do any JSON unmarshaling
 	- handles 301s, etc.
	- handles retries
	- handles timeouts

	it is called with an apiOperation type because the URL it will use varies
	depending on the API operation type (api_QUERY vs. api_WRITE)

 * *****************************************************************/

func (conn *Connection) rqliteApiPost(apiOp apiOperation, sqlStatements []string) ([]byte, error) {
	var responseBody []byte

	switch apiOp {
	case api_QUERY:
		trace("%s: rqliteApiGet() post called for a QUERY of %d statements", conn.ID, len(sqlStatements))
	case api_WRITE:
		trace("%s: rqliteApiGet() post called for a QUERY of %d statements", conn.ID, len(sqlStatements))
	default:
		return responseBody, errors.New("weird! called for an invalid apiOperation in rqliteApiPost()")
	}

	// jsonify the statements.  not really needed in the
	// case of api_STATUS but doesn't hurt

	jStatements, err := json.Marshal(sqlStatements)
	if err != nil {
		return nil, err
	}

	// just to be safe, check this
	peersToTry := conn.cluster.makePeerList()
	if len(peersToTry) < 1 {
		return responseBody, errors.New("I don't have any cluster info")
	}

	// failure log is used so that if all peers fail, we can say something
	// about why each failed
	failureLog := make([]string, 0)

PeerLoop:
	for peerNum, peer := range peersToTry {
		trace("%s: trying peer #%d", conn.ID, peerNum)

		// we're doing a post, and the RFCs say that if you get a 301, it's not
		// automatically followed, so we have to do that ourselves

		responseStatus := "Haven't Tried Yet"
		var url string
		for responseStatus == "Haven't Tried Yet" || responseStatus == "301 Moved Permanently" {
			url = conn.assembleURL(apiOp, peer)
			req, err := http.NewRequest("POST", url, bytes.NewBuffer(jStatements))
			if err != nil {
				trace("%s: got error '%s' doing http.NewRequest", conn.ID, err.Error())
				failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
				continue PeerLoop
			}
			req.Header.Set("Content-Type", "application/json")
			client := &http.Client{}
			response, err := client.Do(req)
			if err != nil {
				trace("%s: got error '%s' doing client.Do", conn.ID, err.Error())
				failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
				continue PeerLoop
			}
			defer response.Body.Close()
			responseBody, err = ioutil.ReadAll(response.Body)
			if err != nil {
				trace("%s: got error '%s' doing ioutil.ReadAll", conn.ID, err.Error())
				failureLog = append(failureLog, fmt.Sprintf("%s failed due to %s", url, err.Error()))
				continue PeerLoop
			}
			responseStatus = response.Status
			if responseStatus == "301 Moved Permanently" {
				v := response.Header["Location"]
				failureLog = append(failureLog, fmt.Sprintf("%s redirected me to %s", url, v[0]))
				url = v[0]
				continue PeerLoop
			} else if responseStatus == "200 OK" {
				trace("%s: api call OK, returning", conn.ID)
				return responseBody, nil
			} else {
				trace("%s: got error in responseStatus: %s", conn.ID, responseStatus)
				failureLog = append(failureLog, fmt.Sprintf("%s failed, got: %s", url, response.Status))
				continue PeerLoop
			}
		}
	}

	// if we got here, all peers failed.  Let's build a verbose error message
	var stringBuffer bytes.Buffer
	stringBuffer.WriteString("tried all peers unsuccessfully. here are the results:\n")
	for n, v := range failureLog {
		stringBuffer.WriteString(fmt.Sprintf("   peer #%d: %s\n", n, v))
	}
	return responseBody, errors.New(stringBuffer.String())
}