X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d005c38bea17107e7be3c5338ad31ba54085df61..908261de15c98553a4200f01aff1f26ef57c8fb8:/services/datamanager/datamanager.go diff --git a/services/datamanager/datamanager.go b/services/datamanager/datamanager.go index ba7446e754..604a6db2f4 100644 --- a/services/datamanager/datamanager.go +++ b/services/datamanager/datamanager.go @@ -108,10 +108,9 @@ func singlerun(arv arvadosclient.ArvadosClient) error { dataFetcher = BuildDataFetcher(arv) } - dataFetcher(arvLogger, &readCollections, &keepServerInfo) - - if readCollections.Err != nil { - return readCollections.Err + err = dataFetcher(arvLogger, &readCollections, &keepServerInfo) + if err != nil { + return err } err = summary.MaybeWriteData(arvLogger, readCollections, keepServerInfo) @@ -138,6 +137,11 @@ func singlerun(arv arvadosclient.ArvadosClient) error { rlbss.Count) } + kc, err := keepclient.MakeKeepClient(&arv) + if err != nil { + return fmt.Errorf("Error setting up keep client %v", 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 { @@ -151,12 +155,6 @@ func singlerun(arv arvadosclient.ArvadosClient) error { }) } - // Not dry-run; issue changes to keepstore - kc, err := keepclient.MakeKeepClient(&arv) - if err != nil { - return fmt.Errorf("Error setting up keep client %v", err.Error()) - } - pullServers := summary.ComputePullServers(kc, &keepServerInfo, readCollections.BlockToDesiredReplication, @@ -183,30 +181,35 @@ func singlerun(arv arvadosclient.ArvadosClient) error { // BuildDataFetcher returns a data fetcher that fetches data from remote servers. func BuildDataFetcher(arv arvadosclient.ArvadosClient) summary.DataFetcher { - return func(arvLogger *logger.Logger, + return func( + arvLogger *logger.Logger, readCollections *collection.ReadCollections, - keepServerInfo *keep.ReadServers) { - collectionChannel := make(chan collection.ReadCollections) - + keepServerInfo *keep.ReadServers, + ) error { + collDone := make(chan struct{}) + var collErr error go func() { - collectionChannel <- collection.GetCollectionsAndSummarize( + *readCollections, collErr = collection.GetCollectionsAndSummarize( collection.GetCollectionsParams{ Client: arv, Logger: arvLogger, BatchSize: 50}) + collDone <- struct{}{} }() - var err error - *keepServerInfo, err = keep.GetKeepServersAndSummarize( + var keepErr error + *keepServerInfo, keepErr = keep.GetKeepServersAndSummarize( keep.GetKeepServersParams{ Client: arv, Logger: arvLogger, Limit: 1000}) - if err != nil { - return - } + <- collDone - *readCollections = <-collectionChannel + // Return a nil error only if both parts succeeded. + if collErr != nil { + return collErr + } + return keepErr } }