func main() {
flag.Parse()
if minutesBetweenRuns == 0 {
- singlerun()
+ err := singlerun()
+ if err != nil {
+ log.Fatalf("Got an error: %v", err)
+ }
} else {
waitTime := time.Minute * time.Duration(minutesBetweenRuns)
for {
log.Println("Beginning Run")
- singlerun()
+ err := singlerun()
+ if err != nil {
+ log.Printf("Got an error: %v", err)
+ }
log.Printf("Sleeping for %d minutes", minutesBetweenRuns)
time.Sleep(waitTime)
}
}
}
-func singlerun() {
+func singlerun() error {
arv, err := arvadosclient.MakeArvadosClient()
if err != nil {
log.Fatalf("Error setting up arvados client %s", err.Error())
fmt.Sprintf("Error setting up keep client %s", err.Error()))
}
+ // Log that we're finished. We force the recording, since go will
+ // not wait for the write timer before exiting.
+ if arvLogger != nil {
+ defer arvLogger.FinalUpdate(func(p map[string]interface{}, e map[string]interface{}) {
+ summaryInfo := logger.GetOrCreateMap(p, "summary_info")
+ summaryInfo["block_replication_counts"] = bucketCounts
+ summaryInfo["replication_summary"] = replicationCounts
+ p["summary_info"] = summaryInfo
+
+ p["run_info"].(map[string]interface{})["finished_at"] = time.Now()
+ })
+ }
+
pullServers := summary.ComputePullServers(kc,
&keepServerInfo,
readCollections.BlockToDesiredReplication,
replicationSummary.UnderReplicatedBlocks)
pullLists := summary.BuildPullLists(pullServers)
- trashLists := summary.BuildTrashLists(kc,
+
+ trashLists, trashErr := summary.BuildTrashLists(kc,
&keepServerInfo,
replicationSummary.KeepBlocksNotInCollections)
summary.WritePullLists(arvLogger, pullLists)
- keep.SendTrashLists(arvLogger, kc, trashLists)
- // Log that we're finished. We force the recording, since go will
- // not wait for the write timer before exiting.
- if arvLogger != nil {
- arvLogger.FinalUpdate(func(p map[string]interface{}, e map[string]interface{}) {
- summaryInfo := logger.GetOrCreateMap(p, "summary_info")
- summaryInfo["block_replication_counts"] = bucketCounts
- summaryInfo["replication_summary"] = replicationCounts
- p["summary_info"] = summaryInfo
-
- p["run_info"].(map[string]interface{})["finished_at"] = time.Now()
- })
+ if trashErr != nil {
+ return err
+ } else {
+ keep.SendTrashLists(keep.GetDataManagerToken(arvLogger), kc, trashLists)
}
+
+ return nil
}
// Returns a data fetcher that fetches data from remote servers.
}
}
-func getDataManagerToken(arvLogger *logger.Logger) string {
+func GetDataManagerToken(arvLogger *logger.Logger) string {
readDataManagerToken := func() {
if dataManagerTokenFile == "" {
flag.Usage()
}
req.Header.Add("Authorization",
- fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger)))
+ fmt.Sprintf("OAuth2 %s", GetDataManagerToken(arvLogger)))
return
}
type TrashList []TrashRequest
-func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map[string]TrashList) {
+func SendTrashLists(dataManagerToken string, kc *keepclient.KeepClient, spl map[string]TrashList) {
count := 0
- rendezvous := make(chan bool)
+ barrier := make(chan bool)
+
+ client := kc.Client
for url, v := range spl {
count += 1
go (func(url string, v TrashList) {
defer (func() {
- rendezvous <- true
+ barrier <- true
})()
pipeReader, pipeWriter := io.Pipe()
}
// Add api token header
- req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger)))
+ req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", dataManagerToken))
// Make the request
var resp *http.Response
- if resp, err = kc.Client.Do(req); err != nil {
+ if resp, err = client.Do(req); err != nil {
log.Printf("Error sending trash list to %v error: %v", url, err.Error())
return
}
- log.Printf("Sent trash list to %v: response was HTTP %d", url, resp.Status)
+ log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status)
io.Copy(ioutil.Discard, resp.Body)
resp.Body.Close()
}
for i := 0; i < count; i += 1 {
- <-rendezvous
+ <-barrier
}
}
--- /dev/null
+package keep
+
+import (
+ "encoding/json"
+ "git.curoverse.com/arvados.git/sdk/go/keepclient"
+ . "gopkg.in/check.v1"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+)
+
+// Gocheck boilerplate
+func Test(t *testing.T) {
+ TestingT(t)
+}
+
+type KeepSuite struct{}
+
+var _ = Suite(&KeepSuite{})
+
+type TestHandler struct {
+ request TrashList
+}
+
+func (this *TestHandler) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
+ r := json.NewDecoder(req.Body)
+ r.Decode(&this.request)
+}
+
+func (s *KeepSuite) TestSendTrashLists(c *C) {
+ th := TestHandler{}
+ server := httptest.NewServer(&th)
+
+ tl := map[string]TrashList{
+ server.URL: TrashList{TrashRequest{"000000000000000000000000deadbeef", 99}}}
+
+ kc := keepclient.KeepClient{Client: &http.Client{}}
+ kc.SetServiceRoots(map[string]string{"xxxx": server.URL},
+ map[string]string{"xxxx": server.URL},
+ map[string]string{})
+
+ SendTrashLists("", &kc, tl)
+ server.Close()
+
+ c.Check(th.request,
+ DeepEquals,
+ tl[server.URL])
+
+}
+
+type TestHandlerError struct {
+}
+
+func (this *TestHandlerError) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
+ http.Error(writer, "I'm a teapot", 405)
+}
+
+func (s *KeepSuite) TestSendTrashListError(c *C) {
+ // Server responds with an error
+
+ th := TestHandlerError{}
+ server := httptest.NewServer(&th)
+
+ tl := map[string]TrashList{
+ server.URL: TrashList{TrashRequest{"000000000000000000000000deadbeef", 99}}}
+
+ kc := keepclient.KeepClient{Client: &http.Client{}}
+ kc.SetServiceRoots(map[string]string{"xxxx": server.URL},
+ map[string]string{"xxxx": server.URL},
+ map[string]string{})
+
+ SendTrashLists("", &kc, tl)
+ server.Close()
+}
+
+func (s *KeepSuite) TestSendTrashListError2(c *C) {
+ // Server is not reachable
+
+ th := TestHandler{}
+ server := httptest.NewServer(&th)
+ server.Close()
+
+ tl := map[string]TrashList{
+ server.URL: TrashList{TrashRequest{"000000000000000000000000deadbeef", 99}}}
+
+ kc := keepclient.KeepClient{Client: &http.Client{}}
+ kc.SetServiceRoots(map[string]string{"xxxx": server.URL},
+ map[string]string{"xxxx": server.URL},
+ map[string]string{})
+
+ SendTrashLists("", &kc, tl)
+}
)
// Gocheck boilerplate
-func Test(t *testing.T) {
+func TestPullLists(t *testing.T) {
TestingT(t)
}
-type MySuite struct{}
+type PullSuite struct{}
-var _ = Suite(&MySuite{})
+var _ = Suite(&PullSuite{})
// Helper method to declare string sets more succinctly
// Could be placed somewhere more general.
return
}
-func (s *MySuite) TestPullListPrintsJSONCorrectly(c *C) {
+func (s *PullSuite) TestPullListPrintsJSONCorrectly(c *C) {
pl := PullList{PullRequest{
Locator: Locator(blockdigest.MakeTestDigestSpecifySize(0xBadBeef, 56789)),
Servers: []string{"keep0.qr1hi.arvadosapi.com:25107",
c.Check(string(b), Equals, expectedOutput)
}
-func (s *MySuite) TestCreatePullServers(c *C) {
+func (s *PullSuite) TestCreatePullServers(c *C) {
var cs CanonicalString
c.Check(
CreatePullServers(cs,
Params: []string{"obtained", "expected"},
}}
-func (s *MySuite) TestBuildPullLists(c *C) {
+func (s *PullSuite) TestBuildPullLists(c *C) {
c.Check(
BuildPullLists(map[Locator]PullServers{}),
PullListMapEquals,
package summary
import (
+ "errors"
+ "fmt"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
"git.curoverse.com/arvados.git/services/datamanager/keep"
- "log"
"time"
)
func BuildTrashLists(kc *keepclient.KeepClient,
keepServerInfo *keep.ReadServers,
- keepBlocksNotInCollections BlockSet) (m map[string]keep.TrashList) {
+ keepBlocksNotInCollections BlockSet) (m map[string]keep.TrashList, err error) {
// Servers that are writeable
writableServers := map[string]struct{}{}
_ttl, err := kc.Arvados.Discovery("blobSignatureTtl")
if err != nil {
- log.Printf("Failed to get blobSignatureTtl: %v", err)
- return map[string]keep.TrashList{}
+ return nil, errors.New(fmt.Sprintf("Failed to get blobSignatureTtl, can't build trash lists: %v", err))
}
ttl := int64(_ttl.(float64))
// expire unreferenced blocks more than "ttl" seconds old.
expiry := time.Now().UTC().Unix() - ttl
- return BuildTrashListsInternal(writableServers, keepServerInfo, expiry, keepBlocksNotInCollections)
+ return buildTrashListsInternal(writableServers, keepServerInfo, expiry, keepBlocksNotInCollections), nil
}
-func BuildTrashListsInternal(writableServers map[string]struct{},
+func buildTrashListsInternal(writableServers map[string]struct{},
keepServerInfo *keep.ReadServers,
expiry int64,
keepBlocksNotInCollections BlockSet) (m map[string]keep.TrashList) {
m = make(map[string]keep.TrashList)
- for block, _ := range keepBlocksNotInCollections {
+ for block := range keepBlocksNotInCollections {
for _, block_on_server := range keepServerInfo.BlockToServers[block] {
- if block_on_server.Mtime < expiry {
- // block is older than expire cutoff
- srv := keepServerInfo.KeepServerIndexToAddress[block_on_server.ServerIndex].String()
+ if block_on_server.Mtime >= expiry {
+ continue
+ }
- _, writable := writableServers[srv]
+ // block is older than expire cutoff
+ srv := keepServerInfo.KeepServerIndexToAddress[block_on_server.ServerIndex].String()
- if writable {
- m[srv] = append(m[srv], keep.TrashRequest{Locator: block.Digest.String(), BlockMtime: block_on_server.Mtime})
- }
+ if _, writable := writableServers[srv]; !writable {
+ continue
}
+
+ m[srv] = append(m[srv], keep.TrashRequest{Locator: block.Digest.String(), BlockMtime: block_on_server.Mtime})
}
}
return
)
// Gocheck boilerplate
-func TrashTest(t *testing.T) {
+func TestTrash(t *testing.T) {
TestingT(t)
}
bs[block0] = struct{}{}
// Test trash list where only sv0 is on writable list.
- c.Check(BuildTrashListsInternal(
+ c.Check(buildTrashListsInternal(
map[string]struct{}{
sv0.URL(): struct{}{}},
&keepServerInfo,
"http://keep0.example.com:80": keep.TrashList{keep.TrashRequest{"000000000000000000000000deadbeef", 99}}})
// Test trash list where both sv0 and sv1 are on writable list.
- c.Check(BuildTrashListsInternal(
+ c.Check(buildTrashListsInternal(
map[string]struct{}{
sv0.URL(): struct{}{},
sv1.URL(): struct{}{}},
"http://keep1.example.com:80": keep.TrashList{keep.TrashRequest{"000000000000000000000000deadbeef", 101}}})
// Test trash list where only block on sv0 is expired
- c.Check(BuildTrashListsInternal(
+ c.Check(buildTrashListsInternal(
map[string]struct{}{
sv0.URL(): struct{}{},
sv1.URL(): struct{}{}},
continue
}
if trashRequest.BlockMtime != mtime.Unix() {
- log.Printf("%v Delete(%v): mtime does not match", volume, trashRequest.Locator)
+ log.Printf("%v Delete(%v): mtime on volume is %v does not match trash list value %v", volume, trashRequest.Locator, mtime.Unix(), trashRequest.BlockMtime)
continue
}
projects / arvados.git / commitdiff