func (h *HTTPApi) listShardInstance(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	db := h.storageNode.Datasources[ps.ByName("datasource")].GetMeta().Databases[ps.ByName("dbname")]
	if bytes, err := json.Marshal(db.ShardInstances); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	} else {
		w.Header().Set("Content-Type", "application/json")
		w.Write(bytes)
	}
}
List all databases that we have in the metadata store
package storagenode
import (
	"encoding/json"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/pprof"
	"time"
	"github.com/jacksontj/dataman/stream/httpjson"
	"github.com/jacksontj/dataman/metrics/promhandler"
	"github.com/julienschmidt/httprouter"
	"github.com/jacksontj/dataman/httputil"
	"github.com/jacksontj/dataman/query"
	"github.com/jacksontj/dataman/storagenode/metadata"
)
type HTTPApi struct {
	storageNode *StorageNode
}
func NewHTTPApi(storageNode *StorageNode) *HTTPApi {
	api := &HTTPApi{
		storageNode: storageNode,
	}
	return api
}
func wrapHandler(h http.Handler) httprouter.Handle {
	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
		h.ServeHTTP(w, r)
	}
}
func (h *HTTPApi) Start(router *httprouter.Router) {
	router.GET("/v1/datasource_instance", httputil.LoggingHandler(h.listDatasourceInstance))
	router.GET("/v1/datasource_instance/:datasource/metadata", httputil.LoggingHandler(h.showMetadata))
	router.GET("/v1/datasource_instance/:datasource/database", httputil.LoggingHandler(h.listDatabase))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname", httputil.LoggingHandler(h.viewDatabase))
	router.POST("/v1/datasource_instance/:datasource/database/:dbname", httputil.LoggingHandler(h.ensureDatabase))
	router.DELETE("/v1/datasource_instance/:datasource/database/:dbname", httputil.LoggingHandler(h.removeDatabase))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance", httputil.LoggingHandler(h.listShardInstance))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance", httputil.LoggingHandler(h.viewShardInstance))
	router.POST("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance", httputil.LoggingHandler(h.ensureShardInstance))
	router.DELETE("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance", httputil.LoggingHandler(h.removeShardInstance))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection", httputil.LoggingHandler(h.listCollection))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname", httputil.LoggingHandler(h.viewCollection))
	router.POST("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname", httputil.LoggingHandler(h.ensureCollection))
	router.DELETE("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname", httputil.LoggingHandler(h.removeCollection))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname/indexes", httputil.LoggingHandler(h.listIndex))
	router.GET("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname/indexes/:indexname", httputil.LoggingHandler(h.viewIndex))
	router.POST("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname/indexes/:indexname", httputil.LoggingHandler(h.ensureIndex))
	router.DELETE("/v1/datasource_instance/:datasource/database/:dbname/shard_instance/:shardinstance/collection/:collectionname/indexes/:indexname", httputil.LoggingHandler(h.removeIndex))
	router.POST("/v1/datasource_instance/:datasource/data/raw/:qtype", httputil.LoggingHandler(h.rawQueryHandler))
	router.GET("/v1/debug/pprof/", wrapHandler(http.HandlerFunc(pprof.Index)))
	router.GET("/v1/debug/pprof/cmdline", wrapHandler(http.HandlerFunc(pprof.Cmdline)))
	router.GET("/v1/debug/pprof/profile", wrapHandler(http.HandlerFunc(pprof.Profile)))
	router.GET("/v1/debug/pprof/symbol", wrapHandler(http.HandlerFunc(pprof.Symbol)))
	router.GET("/v1/debug/pprof/trace", wrapHandler(http.HandlerFunc(pprof.Trace)))
	router.GET("/metrics", wrapHandler(promhandler.Handler(h.storageNode.registry)))
}
func (h *HTTPApi) listDatasourceInstance(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	datasourceInstances := make([]string, 0, len(h.storageNode.Datasources))
	for k := range h.storageNode.Datasources {
		datasourceInstances = append(datasourceInstances, k)
	}
	if bytes, err := json.Marshal(datasourceInstances); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	} else {
		w.Header().Set("Content-Type", "application/json")
		w.Write(bytes)
	}
}
func (h *HTTPApi) showMetadata(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	meta := h.storageNode.Datasources[ps.ByName("datasource")].GetMeta()
	if bytes, err := json.Marshal(meta); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	} else {
		w.Header().Set("Content-Type", "application/json")
		w.Write(bytes)
	}
}
func (h *HTTPApi) listDatabase(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	dbs := h.storageNode.Datasources[ps.ByName("datasource")].GetMeta().ListDatabases()
	if bytes, err := json.Marshal(dbs); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	} else {
		w.Header().Set("Content-Type", "application/json")
		w.Write(bytes)
	}
}
func (h *HTTPApi) viewDatabase(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	meta := h.storageNode.Datasources[ps.ByName("datasource")].GetMeta()
	if db, ok := meta.Databases[ps.ByName("dbname")]; ok {
		if bytes, err := json.Marshal(db); err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			return
		} else {
			w.Header().Set("Content-Type", "application/json")
			w.Write(bytes)
		}
	} else {
		w.WriteHeader(http.StatusNotFound)
		return
	}
}
func (h *HTTPApi) ensureDatabase(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	defer r.Body.Close()
	bytes, _ := ioutil.ReadAll(r.Body)
	var database metadata.Database
	if err := json.Unmarshal(bytes, &database); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		w.Write([]byte(err.Error()))
		return
	} else {
		if err := h.storageNode.Datasources[ps.ByName("datasource")].EnsureExistsDatabase(r.Context(), &database); err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte(err.Error()))
			return
		}
	}
}
func (h *HTTPApi) removeDatabase(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	dbname := ps.ByName("dbname")
	if err := h.storageNode.Datasources[ps.ByName("datasource")].EnsureDoesntExistDatabase(r.Context(), dbname); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		w.Write([]byte(err.Error()))
	}
}