[ARVADOS] updated: d005c38bea17107e7be3c5338ad31ba54085df61
git at public.curoverse.com
git at public.curoverse.com
Mon Nov 30 16:47:35 EST 2015
Summary of changes:
apps/workbench/test/helpers/download_helper.rb | 4 +-
apps/workbench/test/integration_helper.rb | 44 ++++++++--
sdk/go/blockdigest/blockdigest.go | 10 ---
sdk/go/blockdigest/blockdigest_test.go | 24 ++++--
sdk/go/logger/logger.go | 16 ++--
sdk/go/manifest/manifest_test.go | 24 ++++--
services/api/test/unit/crunch_dispatch_test.rb | 7 +-
services/datamanager/collection/collection.go | 3 -
services/datamanager/datamanager.go | 14 +---
services/datamanager/keep/keep.go | 110 ++++++++++++-------------
services/datamanager/keep/keep_test.go | 4 +-
services/datamanager/loggerutil/loggerutil.go | 1 -
services/datamanager/summary/pull_list.go | 37 ++++++---
13 files changed, 170 insertions(+), 128 deletions(-)
via d005c38bea17107e7be3c5338ad31ba54085df61 (commit)
via b70c66d296babe5925e853ae2a5e6cfc2f250c0d (commit)
via 128c2b5e228e1821384064ec50604a1463c29898 (commit)
via 8a6c7aa5eac21e93d721bf958ca67eb7f216888e (commit)
via f04a6c74663185addff5564b47db0442679de78e (commit)
via efc17768290cc63cf7be7b2bccfc5caa77720846 (commit)
via d88316705b23d1022ee2167c7ed0e5cf5e460cbe (commit)
via 40f063c4f6aa35c641650c80751392292e151acb (commit)
via 6936d2cddf2874a7113159e9adcae8c8e67dc48f (commit)
via a1592e2c6d6e9768a1918f3c59b74d6d19033a55 (commit)
via 8d7f3c9ae0648719781174946ea89c3505aabd22 (commit)
via a5ed26a2a1d01646ac511b925f56484be2e8819a (commit)
via 040a541c74913c01ee3517273a7be30c510cc620 (commit)
from 0ec4051e60f34786bf7cf78f5b07f50796c68235 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit d005c38bea17107e7be3c5338ad31ba54085df61
Author: radhika <radhika at curoverse.com>
Date: Mon Nov 30 16:46:36 2015 -0500
7748: address dryRun in summary.WritePullLists and keep.SendTrashLists
diff --git a/services/datamanager/collection/collection.go b/services/datamanager/collection/collection.go
index 4bc5240..33970d8 100644
--- a/services/datamanager/collection/collection.go
+++ b/services/datamanager/collection/collection.go
@@ -357,9 +357,6 @@ func (readCollections *ReadCollections) Summarize(arvLogger *logger.Logger) {
readCollections.OwnerToCollectionSize
collectionInfo["distinct_blocks_named"] =
len(readCollections.BlockToDesiredReplication)
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
diff --git a/services/datamanager/datamanager.go b/services/datamanager/datamanager.go
index 69956f7..ba7446e 100644
--- a/services/datamanager/datamanager.go
+++ b/services/datamanager/datamanager.go
@@ -15,7 +15,6 @@ import (
"git.curoverse.com/arvados.git/services/datamanager/loggerutil"
"git.curoverse.com/arvados.git/services/datamanager/summary"
"log"
- "os"
"time"
)
@@ -149,16 +148,9 @@ func singlerun(arv arvadosclient.ArvadosClient) error {
p["summary_info"] = summaryInfo
p["run_info"].(map[string]interface{})["finished_at"] = time.Now()
- p["run_info"].(map[string]interface{})["args"] = os.Args
})
}
- // If dry-run, do not issue any changes to keepstore
- if dryRun {
- log.Printf("Datamanager dry-run. Returning without issuing any keepstore updates.")
- return nil
- }
-
// Not dry-run; issue changes to keepstore
kc, err := keepclient.MakeKeepClient(&arv)
if err != nil {
@@ -176,7 +168,7 @@ func singlerun(arv arvadosclient.ArvadosClient) error {
&keepServerInfo,
replicationSummary.KeepBlocksNotInCollections)
- err = summary.WritePullLists(arvLogger, pullLists)
+ err = summary.WritePullLists(arvLogger, pullLists, dryRun)
if err != nil {
return err
}
@@ -184,7 +176,7 @@ func singlerun(arv arvadosclient.ArvadosClient) error {
if trashErr != nil {
return err
}
- keep.SendTrashLists(kc, trashLists)
+ keep.SendTrashLists(arvLogger, kc, trashLists, dryRun)
return nil
}
diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go
index 0271ff4..c49a021 100644
--- a/services/datamanager/keep/keep.go
+++ b/services/datamanager/keep/keep.go
@@ -15,7 +15,6 @@ import (
"io/ioutil"
"log"
"net/http"
- "os"
"strconv"
"strings"
"time"
@@ -246,9 +245,6 @@ func GetServerStatus(arvLogger *logger.Logger,
serverInfo["port"] = keepServer.Port
keepInfo[keepServer.UUID] = serverInfo
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
@@ -276,9 +272,6 @@ func GetServerStatus(arvLogger *logger.Logger,
serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
serverInfo["status_response_processed_at"] = now
serverInfo["status"] = keepStatus
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
@@ -298,9 +291,6 @@ func CreateIndexRequest(arvLogger *logger.Logger,
keepInfo := logger.GetOrCreateMap(p, "keep_info")
serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
serverInfo["index_request_sent_at"] = now
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
@@ -329,9 +319,6 @@ func ReadServerResponse(arvLogger *logger.Logger,
keepInfo := logger.GetOrCreateMap(p, "keep_info")
serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
serverInfo["index_response_received_at"] = now
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
@@ -399,9 +386,6 @@ func ReadServerResponse(arvLogger *logger.Logger,
serverInfo["lines_received"] = numLines
serverInfo["duplicates_seen"] = numDuplicates
serverInfo["size_disagreements_seen"] = numSizeDisagreements
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
resp.Body.Close()
@@ -451,9 +435,6 @@ func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) {
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := logger.GetOrCreateMap(p, "keep_info")
keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers)
-
- runInfo := logger.GetOrCreateMap(p, "run_info")
- runInfo["args"] = os.Args
})
}
}
@@ -468,53 +449,64 @@ type TrashRequest struct {
type TrashList []TrashRequest
// SendTrashLists to trash queue
-func SendTrashLists(kc *keepclient.KeepClient, spl map[string]TrashList) (errs []error) {
+func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map[string]TrashList, dryRun bool) (errs []error) {
count := 0
barrier := make(chan error)
client := kc.Client
for url, v := range spl {
- count++
- log.Printf("Sending trash list to %v", url)
-
- go (func(url string, v TrashList) {
- pipeReader, pipeWriter := io.Pipe()
- go (func() {
- enc := json.NewEncoder(pipeWriter)
- enc.Encode(v)
- pipeWriter.Close()
- })()
-
- req, err := http.NewRequest("PUT", fmt.Sprintf("%s/trash", url), pipeReader)
- if err != nil {
- log.Printf("Error creating trash list request for %v error: %v", url, err.Error())
- barrier <- err
- return
- }
-
- req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
-
- // Make the request
- var resp *http.Response
- if resp, err = client.Do(req); err != nil {
- log.Printf("Error sending trash list to %v error: %v", url, err.Error())
- barrier <- err
- return
- }
-
- log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status)
-
- io.Copy(ioutil.Discard, resp.Body)
- resp.Body.Close()
-
- if resp.StatusCode != 200 {
- barrier <- errors.New(fmt.Sprintf("Got HTTP code %v", resp.StatusCode))
- } else {
- barrier <- nil
+ if dryRun {
+ if arvLogger != nil {
+ for url, v := range spl {
+ arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
+ trashListInfo := logger.GetOrCreateMap(p, "trash_list")
+ trashListInfo["started_at"] = time.Now()
+ trashListInfo[url] = v
+ })
+ }
}
- })(url, v)
-
+ } else {
+ count++
+ log.Printf("Sending trash list to %v", url)
+
+ go (func(url string, v TrashList) {
+ pipeReader, pipeWriter := io.Pipe()
+ go (func() {
+ enc := json.NewEncoder(pipeWriter)
+ enc.Encode(v)
+ pipeWriter.Close()
+ })()
+
+ req, err := http.NewRequest("PUT", fmt.Sprintf("%s/trash", url), pipeReader)
+ if err != nil {
+ log.Printf("Error creating trash list request for %v error: %v", url, err.Error())
+ barrier <- err
+ return
+ }
+
+ req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
+
+ // Make the request
+ var resp *http.Response
+ if resp, err = client.Do(req); err != nil {
+ log.Printf("Error sending trash list to %v error: %v", url, err.Error())
+ barrier <- err
+ return
+ }
+
+ log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status)
+
+ io.Copy(ioutil.Discard, resp.Body)
+ resp.Body.Close()
+
+ if resp.StatusCode != 200 {
+ barrier <- errors.New(fmt.Sprintf("Got HTTP code %v", resp.StatusCode))
+ } else {
+ barrier <- nil
+ }
+ })(url, v)
+ }
}
for i := 0; i < count; i++ {
diff --git a/services/datamanager/keep/keep_test.go b/services/datamanager/keep/keep_test.go
index 0691e64..597ab57 100644
--- a/services/datamanager/keep/keep_test.go
+++ b/services/datamanager/keep/keep_test.go
@@ -51,7 +51,7 @@ func (s *KeepSuite) TestSendTrashLists(c *C) {
map[string]string{"xxxx": server.URL},
map[string]string{})
- err := SendTrashLists(&kc, tl)
+ err := SendTrashLists(nil, &kc, tl, false)
c.Check(err, IsNil)
@@ -78,7 +78,7 @@ func sendTrashListError(c *C, server *httptest.Server) {
map[string]string{"xxxx": server.URL},
map[string]string{})
- err := SendTrashLists(&kc, tl)
+ err := SendTrashLists(nil, &kc, tl, false)
c.Check(err, NotNil)
c.Check(err[0], NotNil)
diff --git a/services/datamanager/loggerutil/loggerutil.go b/services/datamanager/loggerutil/loggerutil.go
index 7c45839..8c655cd 100644
--- a/services/datamanager/loggerutil/loggerutil.go
+++ b/services/datamanager/loggerutil/loggerutil.go
@@ -45,7 +45,6 @@ func FatalWithMessage(arvLogger *logger.Logger, message string) {
p["FATAL"] = message
runInfo := logger.GetOrCreateMap(p, "run_info")
runInfo["finished_at"] = time.Now()
- runInfo["args"] = os.Args
})
}
diff --git a/services/datamanager/summary/pull_list.go b/services/datamanager/summary/pull_list.go
index 107abf6..6058616 100644
--- a/services/datamanager/summary/pull_list.go
+++ b/services/datamanager/summary/pull_list.go
@@ -12,6 +12,7 @@ import (
"log"
"os"
"strings"
+ "time"
)
// Locator is a block digest
@@ -175,22 +176,34 @@ func BuildPullLists(lps map[Locator]PullServers) (spl map[string]PullList) {
// This is just a hack for prototyping, it is not expected to be used
// in production.
func WritePullLists(arvLogger *logger.Logger,
- pullLists map[string]PullList) error {
+ pullLists map[string]PullList,
+ dryRun bool) error {
r := strings.NewReplacer(":", ".")
+
for host, list := range pullLists {
- filename := fmt.Sprintf("pull_list.%s", r.Replace(RemoveProtocolPrefix(host)))
- pullListFile, err := os.Create(filename)
- if err != nil {
- return err
- }
- defer pullListFile.Close()
+ if dryRun {
+ if arvLogger != nil {
+ arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
+ pullListInfo := logger.GetOrCreateMap(p, "pull_list")
+ pullListInfo["started_at"] = time.Now()
+ pullListInfo[host] = list
+ })
+ }
+ } else {
+ filename := fmt.Sprintf("pull_list.%s", r.Replace(RemoveProtocolPrefix(host)))
+ pullListFile, err := os.Create(filename)
+ if err != nil {
+ return err
+ }
+ defer pullListFile.Close()
- enc := json.NewEncoder(pullListFile)
- err = enc.Encode(list)
- if err != nil {
- return err
+ enc := json.NewEncoder(pullListFile)
+ err = enc.Encode(list)
+ if err != nil {
+ return err
+ }
+ log.Printf("Wrote pull list to %s.", filename)
}
- log.Printf("Wrote pull list to %s.", filename)
}
return nil
}
commit b70c66d296babe5925e853ae2a5e6cfc2f250c0d
Merge: 8a6c7aa 128c2b5
Author: radhika <radhika at curoverse.com>
Date: Mon Nov 30 10:40:29 2015 -0500
Merge branch 'master' into 7748-datamanager-dry-run
commit 8a6c7aa5eac21e93d721bf958ca67eb7f216888e
Merge: 0ec4051 efc1776
Author: radhika <radhika at curoverse.com>
Date: Fri Nov 27 11:03:10 2015 -0500
Merge branch 'master' into 7748-datamanager-dry-run
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list