- }
- req.Body = ioutil.NopCloser(postBody)
- return nil
-}
-
-type multiClusterQueryResponseCollector struct {
- mtx sync.Mutex
- responses []interface{}
- errors []error
- kind string
-}
-
-func (c *multiClusterQueryResponseCollector) collectResponse(resp *http.Response,
- requestError error) (newResponse *http.Response, err error) {
- if requestError != nil {
- c.mtx.Lock()
- defer c.mtx.Unlock()
- c.errors = append(c.errors, requestError)
- return nil, nil
- }
- defer resp.Body.Close()
- loadInto := make(map[string]interface{})
- err = json.NewDecoder(resp.Body).Decode(&loadInto)
-
- c.mtx.Lock()
- defer c.mtx.Unlock()
-
- if err == nil {
- if resp.StatusCode != http.StatusOK {
- c.errors = append(c.errors, fmt.Errorf("error %v", loadInto["errors"]))
- } else {
- c.responses = append(c.responses, loadInto["items"].([]interface{})...)
- c.kind = loadInto["kind"].(string)
- }
- } else {
- c.errors = append(c.errors, err)
- }
-
- return nil, nil
-}
-
-func (h *genericFederatedRequestHandler) handleMultiClusterQuery(w http.ResponseWriter, req *http.Request,
- params url.Values, clusterId *string) bool {
-
- var filters [][]interface{}
- err := json.Unmarshal([]byte(params["filters"][0]), &filters)
- if err != nil {
- httpserver.Error(w, err.Error(), http.StatusBadRequest)
- return true
- }
-
- // Split the list of uuids by prefix
- queryClusters := make(map[string][]string)
- if len(filters) == 1 && len(filters[0]) == 3 {
- f1 := filters[0]
- lhs := f1[0].(string)
- if lhs == "uuid" {
- op, ok := f1[1].(string)
- if !ok {
- return false
- }
-
- if op == "in" {
- rhs, ok := f1[2].([]interface{})
- if ok {
- for _, i := range rhs {
- u := i.(string)
- *clusterId = u[0:5]
- queryClusters[u[0:5]] = append(queryClusters[u[0:5]], u)
- }
- }
- } else if op == "=" {
- u, ok := f1[2].(string)
- if ok {
- *clusterId = u[0:5]
- queryClusters[u[0:5]] = append(queryClusters[u[0:5]], u)
- }
- }
- }
-
- }
-
- if len(queryClusters) <= 1 {
- // Didn't find ["uuid", "in", ...] filters for multiple clusters
- return false
- }
-
- if !(len(params["count"]) == 1 && (params["count"][0] == `none` ||
- params["count"][0] == `"none"`)) {
- httpserver.Error(w, "Federated multi-object query must have count=none", http.StatusBadRequest)
- return true
- }
-
- wg := sync.WaitGroup{}
-
- // use channel as a semaphore to limit it to 4
- // parallel requests at a time
- sem := make(chan bool, 4)
- defer close(sem)
- req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-
- rc := multiClusterQueryResponseCollector{}
- for k, v := range queryClusters {
- // blocks until it can put a value into the
- // channel (which has a max queue capacity)
- sem <- true
- wg.Add(1)
- go func(k string, v []string) {
- defer func() {
- wg.Done()
- <-sem
- }()
- var remoteReq http.Request
- remoteReq.Header = req.Header
- remoteReq.Method = "POST"
- remoteReq.URL = &url.URL{Path: req.URL.Path}
- remoteParams := make(url.Values)
- remoteParams["_method"] = []string{"GET"}
- remoteParams["count"] = []string{"none"}
- content, err := json.Marshal(v)
- if err != nil {
- rc.mtx.Lock()
- defer rc.mtx.Unlock()
- rc.errors = append(rc.errors, err)
- return
- }
- remoteParams["filters"] = []string{fmt.Sprintf(`[["uuid", "in", %s]]`, content)}
- enc := remoteParams.Encode()
- remoteReq.Body = ioutil.NopCloser(bytes.NewBufferString(enc))
-
- if k == h.handler.Cluster.ClusterID {
- h.handler.localClusterRequest(w, &remoteReq,
- rc.collectResponse)
- } else {
- h.handler.remoteClusterRequest(k, w, &remoteReq,
- rc.collectResponse)
- }
- }(k, v)
- }
- wg.Wait()
-
- if len(rc.errors) > 0 {
- // parallel query
- var strerr []string
- for _, e := range rc.errors {
- strerr = append(strerr, e.Error())
- }
- httpserver.Errors(w, strerr, http.StatusBadGateway)
- } else {
- w.Header().Set("Content-Type", "application/json")
- w.WriteHeader(http.StatusOK)
- itemList := make(map[string]interface{})
- itemList["items"] = rc.responses
- itemList["kind"] = rc.kind
- json.NewEncoder(w).Encode(itemList)
- }
-
- return true
-}
-
-func (h *genericFederatedRequestHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
- m := h.matcher.FindStringSubmatch(req.URL.Path)
- clusterId := ""
-
- if len(m) > 0 && m[2] != "" {
- clusterId = m[2]
- }
-
- // First, parse the query portion of the URL.
- var params url.Values
- var err error
- if params, err = url.ParseQuery(req.URL.RawQuery); err != nil {
- httpserver.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
-
- // Next, if appropriate, merge in parameters from the form POST body.
- if req.Method == "POST" && req.Header.Get("Content-Type") == "application/x-www-form-urlencoded" {
- if err = loadParamsFromForm(req, params); err != nil {
- httpserver.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- }
-
- // Check if the parameters have an explicit cluster_id
- if len(params["cluster_id"]) == 1 {
- clusterId = params["cluster_id"][0]
- }
-
- // Handle the POST-as-GET special case (workaround for large
- // GET requests that potentially exceed maximum URL length,
- // like multi-object queries where the filter has 100s of
- // items)
- effectiveMethod := req.Method
- if req.Method == "POST" && len(params["_method"]) == 1 {
- effectiveMethod = params["_method"][0]
- }
-
- if effectiveMethod == "GET" && clusterId == "" && len(params["filters"]) == 1 {
- if h.handleMultiClusterQuery(w, req, params, &clusterId) {
- return
- }
- }
-
- if clusterId == "" || clusterId == h.handler.Cluster.ClusterID {
- h.next.ServeHTTP(w, req)
- } else {
- h.handler.remoteClusterRequest(clusterId, w, req, nil)
- }
-}
-
-type rewriteSignaturesClusterId struct {
- clusterID string
- expectHash string
-}
-
-func (rw rewriteSignaturesClusterId) rewriteSignatures(resp *http.Response, requestError error) (newResponse *http.Response, err error) {
- if requestError != nil {
- return resp, requestError
- }
-
- if resp.StatusCode != 200 {
- return resp, nil
- }
-
- originalBody := resp.Body
- defer originalBody.Close()
-
- var col arvados.Collection
- err = json.NewDecoder(resp.Body).Decode(&col)
- if err != nil {
- return nil, err
- }
-
- // rewriting signatures will make manifest text 5-10% bigger so calculate
- // capacity accordingly
- updatedManifest := bytes.NewBuffer(make([]byte, 0, int(float64(len(col.ManifestText))*1.1)))
-
- hasher := md5.New()
- mw := io.MultiWriter(hasher, updatedManifest)
- sz := 0
-
- scanner := bufio.NewScanner(strings.NewReader(col.ManifestText))
- scanner.Buffer(make([]byte, 1048576), len(col.ManifestText))
- for scanner.Scan() {
- line := scanner.Text()
- tokens := strings.Split(line, " ")
- if len(tokens) < 3 {
- return nil, fmt.Errorf("Invalid stream (<3 tokens): %q", line)
- }
-
- n, err := mw.Write([]byte(tokens[0]))
- if err != nil {
- return nil, fmt.Errorf("Error updating manifest: %v", err)
- }
- sz += n
- for _, token := range tokens[1:] {
- n, err = mw.Write([]byte(" "))
- if err != nil {
- return nil, fmt.Errorf("Error updating manifest: %v", err)
- }
- sz += n
-
- m := keepclient.SignedLocatorRe.FindStringSubmatch(token)
- if m != nil {
- // Rewrite the block signature to be a remote signature
- _, err = fmt.Fprintf(updatedManifest, "%s%s%s+R%s-%s%s", m[1], m[2], m[3], rw.clusterID, m[5][2:], m[8])
- if err != nil {
- return nil, fmt.Errorf("Error updating manifest: %v", err)
- }
-
- // for hash checking, ignore signatures
- n, err = fmt.Fprintf(hasher, "%s%s", m[1], m[2])
- if err != nil {
- return nil, fmt.Errorf("Error updating manifest: %v", err)
- }
- sz += n
- } else {
- n, err = mw.Write([]byte(token))
- if err != nil {
- return nil, fmt.Errorf("Error updating manifest: %v", err)
- }
- sz += n
- }