You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

163 lines
4.4 KiB

10 years ago
10 years ago
10 years ago
10 years ago
  1. package topology
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "net/http"
  8. "net/url"
  9. "strconv"
  10. "strings"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/operation"
  13. "github.com/chrislusf/seaweedfs/weed/security"
  14. "github.com/chrislusf/seaweedfs/weed/storage"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. func ReplicatedWrite(masterNode string, s *storage.Store,
  18. volumeId storage.VolumeId, needle *storage.Needle,
  19. r *http.Request) (size uint32, errorStatus string) {
  20. //check JWT
  21. jwt := security.GetJwt(r)
  22. ret, err := s.Write(volumeId, needle)
  23. needToReplicate := !s.HasVolume(volumeId)
  24. if err != nil {
  25. errorStatus = "Failed to write to local disk (" + err.Error() + ")"
  26. } else if ret > 0 {
  27. needToReplicate = needToReplicate || s.GetVolume(volumeId).NeedToReplicate()
  28. } else {
  29. errorStatus = "Failed to write to local disk"
  30. }
  31. if !needToReplicate && ret > 0 {
  32. needToReplicate = s.GetVolume(volumeId).NeedToReplicate()
  33. }
  34. if needToReplicate { //send to other replica locations
  35. if r.FormValue("type") != "replicate" {
  36. if err = distributedOperation(masterNode, s, volumeId, func(location operation.Location) error {
  37. u := url.URL{
  38. Scheme: "http",
  39. Host: location.Url,
  40. Path: r.URL.Path,
  41. }
  42. q := url.Values{
  43. "type": {"replicate"},
  44. }
  45. if needle.LastModified > 0 {
  46. q.Set("ts", strconv.FormatUint(needle.LastModified, 10))
  47. }
  48. if needle.IsChunkedManifest() {
  49. q.Set("cm", "true")
  50. }
  51. u.RawQuery = q.Encode()
  52. pairMap := make(map[string]string)
  53. if needle.HasPairs() {
  54. tmpMap := make(map[string]string)
  55. err := json.Unmarshal(needle.Pairs, &tmpMap)
  56. if err != nil {
  57. glog.V(0).Infoln("Unmarshal pairs error:", err)
  58. }
  59. for k, v := range tmpMap {
  60. pairMap[storage.PairNamePrefix+k] = v
  61. }
  62. }
  63. _, err := operation.Upload(u.String(),
  64. string(needle.Name), bytes.NewReader(needle.Data), needle.IsGzipped(), string(needle.Mime),
  65. pairMap, jwt)
  66. return err
  67. }); err != nil {
  68. ret = 0
  69. errorStatus = fmt.Sprintf("Failed to write to replicas for volume %d: %v", volumeId, err)
  70. }
  71. }
  72. }
  73. size = ret
  74. return
  75. }
  76. func ReplicatedDelete(masterNode string, store *storage.Store,
  77. volumeId storage.VolumeId, n *storage.Needle,
  78. r *http.Request) (uint32, error) {
  79. //check JWT
  80. jwt := security.GetJwt(r)
  81. ret, err := store.Delete(volumeId, n)
  82. if err != nil {
  83. glog.V(0).Infoln("delete error:", err)
  84. return ret, err
  85. }
  86. needToReplicate := !store.HasVolume(volumeId)
  87. if !needToReplicate && ret > 0 {
  88. needToReplicate = store.GetVolume(volumeId).NeedToReplicate()
  89. }
  90. if needToReplicate { //send to other replica locations
  91. if r.FormValue("type") != "replicate" {
  92. if err = distributedOperation(masterNode, store, volumeId, func(location operation.Location) error {
  93. return util.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", jwt)
  94. }); err != nil {
  95. ret = 0
  96. }
  97. }
  98. }
  99. return ret, err
  100. }
  101. type DistributedOperationResult map[string]error
  102. func (dr DistributedOperationResult) Error() error {
  103. var errs []string
  104. for k, v := range dr {
  105. if v != nil {
  106. errs = append(errs, fmt.Sprintf("[%s]: %v", k, v))
  107. }
  108. }
  109. if len(errs) == 0 {
  110. return nil
  111. }
  112. return errors.New(strings.Join(errs, "\n"))
  113. }
  114. type RemoteResult struct {
  115. Host string
  116. Error error
  117. }
  118. func distributedOperation(masterNode string, store *storage.Store, volumeId storage.VolumeId, op func(location operation.Location) error) error {
  119. if lookupResult, lookupErr := operation.Lookup(masterNode, volumeId.String()); lookupErr == nil {
  120. length := 0
  121. selfUrl := (store.Ip + ":" + strconv.Itoa(store.Port))
  122. results := make(chan RemoteResult)
  123. for _, location := range lookupResult.Locations {
  124. if location.Url != selfUrl {
  125. length++
  126. go func(location operation.Location, results chan RemoteResult) {
  127. results <- RemoteResult{location.Url, op(location)}
  128. }(location, results)
  129. }
  130. }
  131. ret := DistributedOperationResult(make(map[string]error))
  132. for i := 0; i < length; i++ {
  133. result := <-results
  134. ret[result.Host] = result.Error
  135. }
  136. if volume := store.GetVolume(volumeId); volume != nil {
  137. if length+1 < volume.ReplicaPlacement.GetCopyCount() {
  138. return fmt.Errorf("replicating opetations [%d] is less than volume's replication copy count [%d]", length+1, volume.ReplicaPlacement.GetCopyCount())
  139. }
  140. }
  141. return ret.Error()
  142. } else {
  143. glog.V(0).Infoln()
  144. return fmt.Errorf("Failed to lookup for %d: %v", volumeId, lookupErr)
  145. }
  146. return nil
  147. }