store_replicate.go 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196
  1. package topology
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "fmt"
  6. "net/http"
  7. "net/url"
  8. "strconv"
  9. "strings"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/operation"
  12. "github.com/chrislusf/seaweedfs/weed/security"
  13. "github.com/chrislusf/seaweedfs/weed/storage"
  14. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  15. "github.com/chrislusf/seaweedfs/weed/storage/types"
  16. "github.com/chrislusf/seaweedfs/weed/util"
  17. )
  18. func ReplicatedWrite(masterFn operation.GetMasterFn, s *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request) (isUnchanged bool, err error) {
  19. //check JWT
  20. jwt := security.GetJwt(r)
  21. // check whether this is a replicated write request
  22. var remoteLocations []operation.Location
  23. if r.FormValue("type") != "replicate" {
  24. // this is the initial request
  25. remoteLocations, err = getWritableRemoteReplications(s, volumeId, masterFn)
  26. if err != nil {
  27. glog.V(0).Infoln(err)
  28. return
  29. }
  30. }
  31. // read fsync value
  32. fsync := false
  33. if r.FormValue("fsync") == "true" {
  34. fsync = true
  35. }
  36. if s.GetVolume(volumeId) != nil {
  37. isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, fsync)
  38. if err != nil {
  39. err = fmt.Errorf("failed to write to local disk: %v", err)
  40. glog.V(0).Infoln(err)
  41. return
  42. }
  43. }
  44. if len(remoteLocations) > 0 { //send to other replica locations
  45. if err = distributedOperation(remoteLocations, s, func(location operation.Location) error {
  46. u := url.URL{
  47. Scheme: "http",
  48. Host: location.Url,
  49. Path: r.URL.Path,
  50. }
  51. q := url.Values{
  52. "type": {"replicate"},
  53. "ttl": {n.Ttl.String()},
  54. }
  55. if n.LastModified > 0 {
  56. q.Set("ts", strconv.FormatUint(n.LastModified, 10))
  57. }
  58. if n.IsChunkedManifest() {
  59. q.Set("cm", "true")
  60. }
  61. u.RawQuery = q.Encode()
  62. pairMap := make(map[string]string)
  63. if n.HasPairs() {
  64. tmpMap := make(map[string]string)
  65. err := json.Unmarshal(n.Pairs, &tmpMap)
  66. if err != nil {
  67. glog.V(0).Infoln("Unmarshal pairs error:", err)
  68. }
  69. for k, v := range tmpMap {
  70. pairMap[needle.PairNamePrefix+k] = v
  71. }
  72. }
  73. // volume server do not know about encryption
  74. // TODO optimize here to compress data only once
  75. _, err := operation.UploadData(u.String(), string(n.Name), false, n.Data, n.IsCompressed(), string(n.Mime), pairMap, jwt)
  76. return err
  77. }); err != nil {
  78. err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
  79. glog.V(0).Infoln(err)
  80. }
  81. }
  82. return
  83. }
  84. func ReplicatedDelete(masterFn operation.GetMasterFn, store *storage.Store,
  85. volumeId needle.VolumeId, n *needle.Needle,
  86. r *http.Request) (size types.Size, err error) {
  87. //check JWT
  88. jwt := security.GetJwt(r)
  89. var remoteLocations []operation.Location
  90. if r.FormValue("type") != "replicate" {
  91. remoteLocations, err = getWritableRemoteReplications(store, volumeId, masterFn)
  92. if err != nil {
  93. glog.V(0).Infoln(err)
  94. return
  95. }
  96. }
  97. size, err = store.DeleteVolumeNeedle(volumeId, n)
  98. if err != nil {
  99. glog.V(0).Infoln("delete error:", err)
  100. return
  101. }
  102. if len(remoteLocations) > 0 { //send to other replica locations
  103. if err = distributedOperation(remoteLocations, store, func(location operation.Location) error {
  104. return util.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", string(jwt))
  105. }); err != nil {
  106. size = 0
  107. }
  108. }
  109. return
  110. }
  111. type DistributedOperationResult map[string]error
  112. func (dr DistributedOperationResult) Error() error {
  113. var errs []string
  114. for k, v := range dr {
  115. if v != nil {
  116. errs = append(errs, fmt.Sprintf("[%s]: %v", k, v))
  117. }
  118. }
  119. if len(errs) == 0 {
  120. return nil
  121. }
  122. return errors.New(strings.Join(errs, "\n"))
  123. }
  124. type RemoteResult struct {
  125. Host string
  126. Error error
  127. }
  128. func distributedOperation(locations []operation.Location, store *storage.Store, op func(location operation.Location) error) error {
  129. length := len(locations)
  130. results := make(chan RemoteResult)
  131. for _, location := range locations {
  132. go func(location operation.Location, results chan RemoteResult) {
  133. results <- RemoteResult{location.Url, op(location)}
  134. }(location, results)
  135. }
  136. ret := DistributedOperationResult(make(map[string]error))
  137. for i := 0; i < length; i++ {
  138. result := <-results
  139. ret[result.Host] = result.Error
  140. }
  141. return ret.Error()
  142. }
  143. func getWritableRemoteReplications(s *storage.Store, volumeId needle.VolumeId, masterFn operation.GetMasterFn) (
  144. remoteLocations []operation.Location, err error) {
  145. v := s.GetVolume(volumeId)
  146. if v != nil && v.ReplicaPlacement.GetCopyCount() == 1 {
  147. return
  148. }
  149. // not on local store, or has replications
  150. lookupResult, lookupErr := operation.Lookup(masterFn, volumeId.String())
  151. if lookupErr == nil {
  152. selfUrl := s.Ip + ":" + strconv.Itoa(s.Port)
  153. for _, location := range lookupResult.Locations {
  154. if location.Url != selfUrl {
  155. remoteLocations = append(remoteLocations, location)
  156. }
  157. }
  158. } else {
  159. err = fmt.Errorf("failed to lookup for %d: %v", volumeId, lookupErr)
  160. return
  161. }
  162. if v != nil {
  163. // has one local and has remote replications
  164. copyCount := v.ReplicaPlacement.GetCopyCount()
  165. if len(lookupResult.Locations) < copyCount {
  166. err = fmt.Errorf("replicating opetations [%d] is less than volume %d replication copy count [%d]",
  167. len(lookupResult.Locations), volumeId, copyCount)
  168. }
  169. }
  170. return
  171. }