cluster.go 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342
  1. /*
  2. * EliasDB
  3. *
  4. * Copyright 2016 Matthias Ladkau. All rights reserved.
  5. *
  6. * This Source Code Form is subject to the terms of the Mozilla Public
  7. * License, v. 2.0. If a copy of the MPL was not distributed with this
  8. * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  9. */
  10. package v1
  11. import (
  12. "encoding/json"
  13. "fmt"
  14. "net/http"
  15. "devt.de/krotik/eliasdb/api"
  16. )
  17. /*
  18. EndpointClusterQuery is the cluster endpoint URL (rooted). Handles everything under cluster/...
  19. */
  20. const EndpointClusterQuery = api.APIRoot + APIv1 + "/cluster/"
  21. /*
  22. ClusterEndpointInst creates a new endpoint handler.
  23. */
  24. func ClusterEndpointInst() api.RestEndpointHandler {
  25. return &clusterEndpoint{}
  26. }
  27. /*
  28. Handler object for cluster queries.
  29. */
  30. type clusterEndpoint struct {
  31. *api.DefaultEndpointHandler
  32. }
  33. /*
  34. HandleGET handles a cluster query REST call.
  35. */
  36. func (ce *clusterEndpoint) HandleGET(w http.ResponseWriter, r *http.Request, resources []string) {
  37. var data interface{}
  38. // Check clustering is enabled
  39. if api.DD == nil || api.DDLog == nil {
  40. http.Error(w, "Clustering is not enabled on this instance", http.StatusServiceUnavailable)
  41. return
  42. }
  43. if len(resources) == 1 && resources[0] == "log" {
  44. // Cluster logs are requested
  45. data = api.DDLog.StringSlice()
  46. } else if len(resources) == 1 && resources[0] == "memberinfos" {
  47. // Cluster member infos are requested
  48. data = api.DD.MemberManager.MemberInfoCluster()
  49. } else {
  50. // By default the cluster state is returned
  51. data = api.DD.MemberManager.StateInfo().Map()
  52. }
  53. // Write data
  54. w.Header().Set("content-type", "application/json; charset=utf-8")
  55. ret := json.NewEncoder(w)
  56. ret.Encode(data)
  57. }
  58. /*
  59. HandlePUT handles a REST call to join/eject/ping members of the cluster.
  60. */
  61. func (ce *clusterEndpoint) HandlePUT(w http.ResponseWriter, r *http.Request, resources []string) {
  62. // Check parameters
  63. if !checkResources(w, resources, 1, 1, "Need a command either: join or eject") {
  64. return
  65. }
  66. dec := json.NewDecoder(r.Body)
  67. args := make(map[string]string)
  68. if err := dec.Decode(&args); err != nil {
  69. http.Error(w, "Could not decode arguments: "+err.Error(), http.StatusBadRequest)
  70. return
  71. }
  72. // Function to check arguments
  73. getArg := func(arg string) (string, bool) {
  74. v, ok := args[arg]
  75. if !ok {
  76. http.Error(w, fmt.Sprintf("Required argument %v missing in body arguments", arg), http.StatusBadRequest)
  77. }
  78. return v, ok
  79. }
  80. if resources[0] == "join" {
  81. // Get required args
  82. name, ok := getArg("name")
  83. if ok {
  84. rpc, ok := getArg("netaddr")
  85. if ok {
  86. err := api.DD.MemberManager.JoinCluster(name, rpc)
  87. if err != nil {
  88. http.Error(w, "Could not join the cluster: "+err.Error(), http.StatusForbidden)
  89. }
  90. }
  91. }
  92. } else if resources[0] == "eject" {
  93. // Get required args
  94. name, ok := getArg("name")
  95. if ok {
  96. err := api.DD.MemberManager.EjectMember(name)
  97. if err != nil {
  98. http.Error(w, "Could not eject "+name+" from cluster: "+err.Error(), http.StatusForbidden)
  99. }
  100. }
  101. } else if resources[0] == "ping" {
  102. // Get required args
  103. name, ok := getArg("name")
  104. if ok {
  105. rpc, ok := getArg("netaddr")
  106. if ok {
  107. res, err := api.DD.MemberManager.Client.SendPing(name, rpc)
  108. if err != nil {
  109. http.Error(w, "Ping returned an error: "+err.Error(), http.StatusForbidden)
  110. } else {
  111. w.Header().Set("content-type", "application/json; charset=utf-8")
  112. ret := json.NewEncoder(w)
  113. ret.Encode(res)
  114. }
  115. }
  116. }
  117. } else {
  118. http.Error(w, "Unknown command: "+resources[0], http.StatusBadRequest)
  119. }
  120. }
  121. /*
  122. HandleDELETE handles a cluster delete REST call.
  123. */
  124. func (ce *clusterEndpoint) HandleDELETE(w http.ResponseWriter, r *http.Request, resources []string) {
  125. // Check clustering is enabled
  126. if api.DD == nil || api.DDLog == nil {
  127. http.Error(w, "Clustering is not enabled on this instance", http.StatusServiceUnavailable)
  128. return
  129. }
  130. if len(resources) == 1 && resources[0] == "log" {
  131. // Cluster log should be reset
  132. api.DDLog.Reset()
  133. return
  134. }
  135. http.Error(w, "Request had no effect", http.StatusBadRequest)
  136. }
  137. /*
  138. SwaggerDefs is used to describe the endpoint in swagger.
  139. */
  140. func (ce *clusterEndpoint) SwaggerDefs(s map[string]interface{}) {
  141. s["paths"].(map[string]interface{})["/v1/cluster"] = map[string]interface{}{
  142. "get": map[string]interface{}{
  143. "summary": "Return cluster specific information.",
  144. "description": "The cluster endpoint returns the cluster state info which contains cluster members and their state.",
  145. "produces": []string{
  146. "text/plain",
  147. "application/json",
  148. },
  149. "responses": map[string]interface{}{
  150. "200": map[string]interface{}{
  151. "description": "A key-value map.",
  152. },
  153. "default": map[string]interface{}{
  154. "description": "Error response",
  155. "schema": map[string]interface{}{
  156. "$ref": "#/definitions/Error",
  157. },
  158. },
  159. },
  160. },
  161. }
  162. s["paths"].(map[string]interface{})["/v1/cluster/{command}"] = map[string]interface{}{
  163. "put": map[string]interface{}{
  164. "summary": "Commands can be given to the cluster by using PUT requests.",
  165. "description": "The cluster can be controlled via this command endpoint on any member.",
  166. "consumes": []string{
  167. "application/json",
  168. },
  169. "produces": []string{
  170. "text/plain",
  171. "application/json",
  172. },
  173. "parameters": []map[string]interface{}{
  174. map[string]interface{}{
  175. "name": "command",
  176. "in": "path",
  177. "description": "Valid commands are: ping, join and eject.",
  178. "required": true,
  179. "type": "string",
  180. },
  181. map[string]interface{}{
  182. "name": "args",
  183. "in": "body",
  184. "description": "Arguments for a command",
  185. "required": true,
  186. "schema": map[string]interface{}{
  187. "type": "object",
  188. "properties": map[string]interface{}{
  189. "name": map[string]interface{}{
  190. "description": "Name of a cluster member (ping/join=member to contact, eject=member to eject).",
  191. "type": "string",
  192. },
  193. "netaddr": map[string]interface{}{
  194. "description": "Network address of a member e.g. localhost:9030 (ping/join=member address to contact)",
  195. "type": "string",
  196. },
  197. },
  198. },
  199. },
  200. },
  201. "responses": map[string]interface{}{
  202. "200": map[string]interface{}{
  203. "description": "Only the ping command returns its result. All other positive responses are empty.",
  204. },
  205. "default": map[string]interface{}{
  206. "description": "Error response",
  207. "schema": map[string]interface{}{
  208. "$ref": "#/definitions/Error",
  209. },
  210. },
  211. },
  212. },
  213. }
  214. s["paths"].(map[string]interface{})["/v1/cluster/memberinfos"] = map[string]interface{}{
  215. "get": map[string]interface{}{
  216. "summary": "Return static member info of every known cluster member.",
  217. "description": "The memberinfos returns the static member info of every known cluster member. If a member is not reachable its info contains a single key-value pair with the key error and an error message as value.",
  218. "produces": []string{
  219. "text/plain",
  220. "application/json",
  221. },
  222. "responses": map[string]interface{}{
  223. "200": map[string]interface{}{
  224. "description": "A map of memberinfos (keys are member names).",
  225. },
  226. "default": map[string]interface{}{
  227. "description": "Error response",
  228. "schema": map[string]interface{}{
  229. "$ref": "#/definitions/Error",
  230. },
  231. },
  232. },
  233. },
  234. }
  235. s["paths"].(map[string]interface{})["/v1/cluster/log"] = map[string]interface{}{
  236. "get": map[string]interface{}{
  237. "summary": "Return latest cluster related log messages.",
  238. "description": "The cluster log endpoint returns the latest cluster related log messages from a memory ring buffer.",
  239. "produces": []string{
  240. "text/plain",
  241. "application/json",
  242. },
  243. "responses": map[string]interface{}{
  244. "200": map[string]interface{}{
  245. "description": "A list of log messages.",
  246. },
  247. "default": map[string]interface{}{
  248. "description": "Error response",
  249. "schema": map[string]interface{}{
  250. "$ref": "#/definitions/Error",
  251. },
  252. },
  253. },
  254. },
  255. "delete": map[string]interface{}{
  256. "summary": "Reset the cluster log.",
  257. "description": "A delete call to the log endpoint resets the cluster related log and clears the ring buffer in memory.",
  258. "produces": []string{
  259. "text/plain",
  260. "application/json",
  261. },
  262. "responses": map[string]interface{}{
  263. "200": map[string]interface{}{
  264. "description": "Cluster related log was reset.",
  265. },
  266. "default": map[string]interface{}{
  267. "description": "Error response",
  268. "schema": map[string]interface{}{
  269. "$ref": "#/definitions/Error",
  270. },
  271. },
  272. },
  273. },
  274. }
  275. // Add generic error object to definition
  276. s["definitions"].(map[string]interface{})["Error"] = map[string]interface{}{
  277. "description": "A human readable error mesage.",
  278. "type": "string",
  279. }
  280. }