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.

363 lines
10 KiB

12 years ago
12 years ago
12 years ago
  1. package storage
  2. import (
  3. "code.google.com/p/weed-fs/go/glog"
  4. "code.google.com/p/weed-fs/go/operation"
  5. "code.google.com/p/weed-fs/go/util"
  6. "encoding/json"
  7. "errors"
  8. "fmt"
  9. "io/ioutil"
  10. "math/rand"
  11. "net/url"
  12. "strconv"
  13. "strings"
  14. )
  15. type DiskLocation struct {
  16. Directory string
  17. MaxVolumeCount int
  18. volumes map[VolumeId]*Volume
  19. }
  20. func (mn *DiskLocation) reset() {
  21. }
  22. type MasterNodes struct {
  23. nodes []string
  24. lastNode int
  25. }
  26. func NewMasterNodes(bootstrapNode string) (mn *MasterNodes) {
  27. mn = &MasterNodes{nodes: []string{bootstrapNode}, lastNode: -1}
  28. return
  29. }
  30. func (mn *MasterNodes) reset() {
  31. if len(mn.nodes) > 1 && mn.lastNode > 0 {
  32. mn.lastNode = -mn.lastNode
  33. }
  34. }
  35. func (mn *MasterNodes) findMaster() (string, error) {
  36. if len(mn.nodes) == 0 {
  37. return "", errors.New("No master node found!")
  38. }
  39. if mn.lastNode < 0 {
  40. for _, m := range mn.nodes {
  41. if masters, e := operation.ListMasters(m); e == nil {
  42. mn.nodes = masters
  43. mn.lastNode = rand.Intn(len(mn.nodes))
  44. glog.V(2).Info("current master node is :", mn.nodes[mn.lastNode])
  45. break
  46. }
  47. }
  48. }
  49. if mn.lastNode < 0 {
  50. return "", errors.New("No master node avalable!")
  51. }
  52. return mn.nodes[mn.lastNode], nil
  53. }
  54. type Store struct {
  55. Port int
  56. Ip string
  57. PublicUrl string
  58. Locations []*DiskLocation
  59. dataCenter string //optional informaton, overwriting master setting if exists
  60. rack string //optional information, overwriting master setting if exists
  61. connected bool
  62. volumeSizeLimit uint64 //read from the master
  63. masterNodes *MasterNodes
  64. }
  65. func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int) (s *Store) {
  66. s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl}
  67. s.Locations = make([]*DiskLocation, 0)
  68. for i := 0; i < len(dirnames); i++ {
  69. location := &DiskLocation{Directory: dirnames[i], MaxVolumeCount: maxVolumeCounts[i]}
  70. location.volumes = make(map[VolumeId]*Volume)
  71. location.loadExistingVolumes()
  72. s.Locations = append(s.Locations, location)
  73. }
  74. return
  75. }
  76. func (s *Store) AddVolume(volumeListString string, collection string, replicaPlacement string) error {
  77. rt, e := NewReplicaPlacementFromString(replicaPlacement)
  78. if e != nil {
  79. return e
  80. }
  81. for _, range_string := range strings.Split(volumeListString, ",") {
  82. if strings.Index(range_string, "-") < 0 {
  83. id_string := range_string
  84. id, err := NewVolumeId(id_string)
  85. if err != nil {
  86. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", id_string)
  87. }
  88. e = s.addVolume(VolumeId(id), collection, rt)
  89. } else {
  90. pair := strings.Split(range_string, "-")
  91. start, start_err := strconv.ParseUint(pair[0], 10, 64)
  92. if start_err != nil {
  93. return fmt.Errorf("Volume Start Id %s is not a valid unsigned integer!", pair[0])
  94. }
  95. end, end_err := strconv.ParseUint(pair[1], 10, 64)
  96. if end_err != nil {
  97. return fmt.Errorf("Volume End Id %s is not a valid unsigned integer!", pair[1])
  98. }
  99. for id := start; id <= end; id++ {
  100. if err := s.addVolume(VolumeId(id), collection, rt); err != nil {
  101. e = err
  102. }
  103. }
  104. }
  105. }
  106. return e
  107. }
  108. func (s *Store) DeleteCollection(collection string) (e error) {
  109. for _, location := range s.Locations {
  110. for k, v := range location.volumes {
  111. if v.Collection == collection {
  112. e = v.Destroy()
  113. if e != nil {
  114. return
  115. }
  116. delete(location.volumes, k)
  117. }
  118. }
  119. }
  120. return
  121. }
  122. func (s *Store) findVolume(vid VolumeId) *Volume {
  123. for _, location := range s.Locations {
  124. if v, found := location.volumes[vid]; found {
  125. return v
  126. }
  127. }
  128. return nil
  129. }
  130. func (s *Store) findFreeLocation() (ret *DiskLocation) {
  131. max := 0
  132. for _, location := range s.Locations {
  133. currentFreeCount := location.MaxVolumeCount - len(location.volumes)
  134. if currentFreeCount > max {
  135. max = currentFreeCount
  136. ret = location
  137. }
  138. }
  139. return ret
  140. }
  141. func (s *Store) addVolume(vid VolumeId, collection string, replicaPlacement *ReplicaPlacement) error {
  142. if s.findVolume(vid) != nil {
  143. return fmt.Errorf("Volume Id %s already exists!", vid)
  144. }
  145. if location := s.findFreeLocation(); location != nil {
  146. glog.V(0).Infoln("In dir", location.Directory, "adds volume =", vid, ", collection =", collection, ", replicaPlacement =", replicaPlacement)
  147. if volume, err := NewVolume(location.Directory, collection, vid, replicaPlacement); err == nil {
  148. location.volumes[vid] = volume
  149. return nil
  150. } else {
  151. return err
  152. }
  153. }
  154. return fmt.Errorf("No more free space left")
  155. }
  156. func (s *Store) CheckCompactVolume(volumeIdString string, garbageThresholdString string) (error, bool) {
  157. vid, err := NewVolumeId(volumeIdString)
  158. if err != nil {
  159. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString), false
  160. }
  161. garbageThreshold, e := strconv.ParseFloat(garbageThresholdString, 32)
  162. if e != nil {
  163. return fmt.Errorf("garbageThreshold %s is not a valid float number!", garbageThresholdString), false
  164. }
  165. if v := s.findVolume(vid); v != nil {
  166. glog.V(3).Infoln(vid, "garbage level is", v.garbageLevel())
  167. return nil, garbageThreshold < v.garbageLevel()
  168. }
  169. return fmt.Errorf("volume id %s is not found during check compact!", vid), false
  170. }
  171. func (s *Store) CompactVolume(volumeIdString string) error {
  172. vid, err := NewVolumeId(volumeIdString)
  173. if err != nil {
  174. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  175. }
  176. if v := s.findVolume(vid); v != nil {
  177. return v.Compact()
  178. }
  179. return fmt.Errorf("volume id %s is not found during compact!", vid)
  180. }
  181. func (s *Store) CommitCompactVolume(volumeIdString string) error {
  182. vid, err := NewVolumeId(volumeIdString)
  183. if err != nil {
  184. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  185. }
  186. if v := s.findVolume(vid); v != nil {
  187. return v.commitCompact()
  188. }
  189. return fmt.Errorf("volume id %s is not found during commit compact!", vid)
  190. }
  191. func (s *Store) FreezeVolume(volumeIdString string) error {
  192. vid, err := NewVolumeId(volumeIdString)
  193. if err != nil {
  194. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  195. }
  196. if v := s.findVolume(vid); v != nil {
  197. if v.readOnly {
  198. return fmt.Errorf("Volume %s is already read-only", volumeIdString)
  199. }
  200. return v.freeze()
  201. }
  202. return fmt.Errorf("volume id %s is not found during freeze!", vid)
  203. }
  204. func (l *DiskLocation) loadExistingVolumes() {
  205. if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
  206. for _, dir := range dirs {
  207. name := dir.Name()
  208. if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
  209. collection := ""
  210. base := name[:len(name)-len(".dat")]
  211. i := strings.Index(base, "_")
  212. if i > 0 {
  213. collection, base = base[0:i], base[i+1:]
  214. }
  215. if vid, err := NewVolumeId(base); err == nil {
  216. if l.volumes[vid] == nil {
  217. if v, e := NewVolume(l.Directory, collection, vid, nil); e == nil {
  218. l.volumes[vid] = v
  219. glog.V(0).Infoln("data file", l.Directory+"/"+name, "replicaPlacement =", v.ReplicaPlacement, "version =", v.Version(), "size =", v.Size())
  220. }
  221. }
  222. }
  223. }
  224. }
  225. }
  226. glog.V(0).Infoln("Store started on dir:", l.Directory, "with", len(l.volumes), "volumes", "max", l.MaxVolumeCount)
  227. }
  228. func (s *Store) Status() []*VolumeInfo {
  229. var stats []*VolumeInfo
  230. for _, location := range s.Locations {
  231. for k, v := range location.volumes {
  232. s := &VolumeInfo{Id: VolumeId(k), Size: v.ContentSize(),
  233. Collection: v.Collection,
  234. ReplicaPlacement: v.ReplicaPlacement,
  235. Version: v.Version(),
  236. FileCount: v.nm.FileCount(),
  237. DeleteCount: v.nm.DeletedCount(),
  238. DeletedByteCount: v.nm.DeletedSize(),
  239. ReadOnly: v.readOnly}
  240. stats = append(stats, s)
  241. }
  242. }
  243. return stats
  244. }
  245. func (s *Store) SetDataCenter(dataCenter string) {
  246. s.dataCenter = dataCenter
  247. }
  248. func (s *Store) SetRack(rack string) {
  249. s.rack = rack
  250. }
  251. func (s *Store) SetBootstrapMaster(bootstrapMaster string) {
  252. s.masterNodes = NewMasterNodes(bootstrapMaster)
  253. }
  254. func (s *Store) Join() error {
  255. masterNode, e := s.masterNodes.findMaster()
  256. if e != nil {
  257. return e
  258. }
  259. stats := new([]*VolumeInfo)
  260. maxVolumeCount := 0
  261. for _, location := range s.Locations {
  262. maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
  263. for k, v := range location.volumes {
  264. s := &VolumeInfo{Id: VolumeId(k), Size: uint64(v.Size()),
  265. Collection: v.Collection,
  266. ReplicaPlacement: v.ReplicaPlacement,
  267. Version: v.Version(),
  268. FileCount: v.nm.FileCount(),
  269. DeleteCount: v.nm.DeletedCount(),
  270. DeletedByteCount: v.nm.DeletedSize(),
  271. ReadOnly: v.readOnly}
  272. *stats = append(*stats, s)
  273. }
  274. }
  275. bytes, _ := json.Marshal(stats)
  276. values := make(url.Values)
  277. if !s.connected {
  278. values.Add("init", "true")
  279. }
  280. values.Add("port", strconv.Itoa(s.Port))
  281. values.Add("ip", s.Ip)
  282. values.Add("publicUrl", s.PublicUrl)
  283. values.Add("volumes", string(bytes))
  284. values.Add("maxVolumeCount", strconv.Itoa(maxVolumeCount))
  285. values.Add("dataCenter", s.dataCenter)
  286. values.Add("rack", s.rack)
  287. jsonBlob, err := util.Post("http://"+masterNode+"/dir/join", values)
  288. if err != nil {
  289. s.masterNodes.reset()
  290. return err
  291. }
  292. var ret operation.JoinResult
  293. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  294. return err
  295. }
  296. if ret.Error != "" {
  297. return errors.New(ret.Error)
  298. }
  299. s.volumeSizeLimit = ret.VolumeSizeLimit
  300. s.connected = true
  301. return nil
  302. }
  303. func (s *Store) Close() {
  304. for _, location := range s.Locations {
  305. for _, v := range location.volumes {
  306. v.Close()
  307. }
  308. }
  309. }
  310. func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
  311. if v := s.findVolume(i); v != nil {
  312. if v.readOnly {
  313. err = fmt.Errorf("Volume %s is read only!", i)
  314. return
  315. } else {
  316. if MaxPossibleVolumeSize >= v.ContentSize()+uint64(size) {
  317. size, err = v.write(n)
  318. } else {
  319. err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.volumeSizeLimit, v.ContentSize())
  320. }
  321. if s.volumeSizeLimit < v.ContentSize()+3*uint64(size) {
  322. glog.V(0).Infoln("volume", i, "size", v.ContentSize(), "will exceed limit", s.volumeSizeLimit)
  323. if e := s.Join(); e != nil {
  324. glog.V(0).Infoln("error when reporting size:", e)
  325. }
  326. }
  327. }
  328. return
  329. }
  330. glog.V(0).Infoln("volume", i, "not found!")
  331. err = fmt.Errorf("Volume %s not found!", i)
  332. return
  333. }
  334. func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
  335. if v := s.findVolume(i); v != nil && !v.readOnly {
  336. return v.delete(n)
  337. }
  338. return 0, nil
  339. }
  340. func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
  341. if v := s.findVolume(i); v != nil {
  342. return v.read(n)
  343. }
  344. return 0, fmt.Errorf("Volume %v not found!", i)
  345. }
  346. func (s *Store) GetVolume(i VolumeId) *Volume {
  347. return s.findVolume(i)
  348. }
  349. func (s *Store) HasVolume(i VolumeId) bool {
  350. v := s.findVolume(i)
  351. return v != nil
  352. }