+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+
package keepclient
import (
"io/ioutil"
"log"
"math/rand"
- "net"
"net/http"
"os"
- "regexp"
"strings"
- "time"
- "git.curoverse.com/arvados.git/sdk/go/streamer"
+ "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
)
// Function used to emit debug messages. The easiest way to enable
var DebugPrintf = func(string, ...interface{}) {}
func init() {
- var matchTrue = regexp.MustCompile("^(?i:1|yes|true)$")
- if matchTrue.MatchString(os.Getenv("ARVADOS_DEBUG")) {
+ if arvadosclient.StringBool(os.Getenv("ARVADOS_DEBUG")) {
DebugPrintf = log.Printf
}
}
return fmt.Sprintf("%x", md5.Sum([]byte(s)))
}
-// Set timeouts applicable when connecting to non-disk services
-// (assumed to be over the Internet).
-func (*KeepClient) setClientSettingsNonDisk(client *http.Client) {
- // Maximum time to wait for a complete response
- client.Timeout = 300 * time.Second
-
- // TCP and TLS connection settings
- client.Transport = &http.Transport{
- Dial: (&net.Dialer{
- // The maximum time to wait to set up
- // the initial TCP connection.
- Timeout: 30 * time.Second,
-
- // The TCP keep alive heartbeat
- // interval.
- KeepAlive: 120 * time.Second,
- }).Dial,
-
- TLSHandshakeTimeout: 10 * time.Second,
- }
-}
-
-// Set timeouts applicable when connecting to keepstore services directly
-// (assumed to be on the local network).
-func (*KeepClient) setClientSettingsDisk(client *http.Client) {
- // Maximum time to wait for a complete response
- client.Timeout = 20 * time.Second
-
- // TCP and TLS connection timeouts
- client.Transport = &http.Transport{
- Dial: (&net.Dialer{
- // The maximum time to wait to set up
- // the initial TCP connection.
- Timeout: 2 * time.Second,
-
- // The TCP keep alive heartbeat
- // interval.
- KeepAlive: 180 * time.Second,
- }).Dial,
-
- TLSHandshakeTimeout: 4 * time.Second,
- }
-}
-
type svcList struct {
Items []keepService `json:"items"`
}
response string
}
-func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser,
+func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.Reader,
upload_status chan<- uploadStatus, expectedLength int64, requestID int32) {
var req *http.Request
if req, err = http.NewRequest("PUT", url, nil); err != nil {
DebugPrintf("DEBUG: [%08x] Error creating request PUT %v error: %v", requestID, url, err.Error())
upload_status <- uploadStatus{err, url, 0, 0, ""}
- body.Close()
return
}
req.ContentLength = expectedLength
if expectedLength > 0 {
- // http.Client.Do will close the body ReadCloser when it is
- // done with it.
- req.Body = body
+ req.Body = ioutil.NopCloser(body)
} else {
- // "For client requests, a value of 0 means unknown if Body is
- // not nil." In this case we do want the body to be empty, so
- // don't set req.Body. However, we still need to close the
- // body ReadCloser.
- body.Close()
+ // "For client requests, a value of 0 means unknown if
+ // Body is not nil." In this case we do want the body
+ // to be empty, so don't set req.Body.
}
req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.Arvados.ApiToken))
req.Header.Add(X_Keep_Desired_Replicas, fmt.Sprint(this.Want_replicas))
var resp *http.Response
- if resp, err = this.Client.Do(req); err != nil {
+ if resp, err = this.httpClient().Do(req); err != nil {
DebugPrintf("DEBUG: [%08x] Upload failed %v error: %v", requestID, url, err.Error())
upload_status <- uploadStatus{err, url, 0, 0, ""}
return
func (this *KeepClient) putReplicas(
hash string,
- tr *streamer.AsyncStream,
+ getReader func() io.Reader,
expectedLength int64) (locator string, replicas int, err error) {
// Generate an arbitrary ID to identify this specific
// Start some upload requests
if next_server < len(sv) {
DebugPrintf("DEBUG: [%08x] Begin upload %s to %s", requestID, hash, sv[next_server])
- go this.uploadToKeepServer(sv[next_server], hash, tr.MakeStreamReader(), upload_status, expectedLength, requestID)
+ go this.uploadToKeepServer(sv[next_server], hash, getReader(), upload_status, expectedLength, requestID)
next_server += 1
active += 1
} else {