my-rook-config/pkg/api/object.go

/*
Copyright 2016 The Rook Authors. All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package api

import (
	"encoding/json"
	"net/http"
	"sort"

	"github.com/gorilla/mux"
	"github.com/rook/rook/pkg/ceph/rgw"
	"github.com/rook/rook/pkg/model"
)

const (
	defaultObjectStoreName = "default"
	defaultRGWInstances    = 1
)

func (h *Handler) objectContext(r *http.Request) *rgw.Context {
	storeName := defaultObjectStoreName
	if name, ok := mux.Vars(r)["name"]; ok {
		storeName = name
	}

	var clusterName string
	info, err := h.config.ClusterHandler.GetClusterInfo()
	if err != nil {
		logger.Infof("defaulting the object context to the rook namespace")
		clusterName = "rook"
	} else {
		clusterName = info.Name
	}

	return rgw.NewContext(h.context, storeName, clusterName)
}

// GetObjectStores gets the object stores in this cluster.
// GET
// /objectstore
func (h *Handler) GetObjectStores(w http.ResponseWriter, r *http.Request) {
	stores, err := h.config.ClusterHandler.GetObjectStores()
	if err != nil {
		logger.Errorf("failed to get object stores: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	FormatJsonResponse(w, stores)
}

// CreateObjectStore creates a new object store in this cluster.
// POST
// /objectstore
func (h *Handler) CreateObjectStore(w http.ResponseWriter, r *http.Request) {
	var objectStore model.ObjectStore
	err := json.NewDecoder(r.Body).Decode(&objectStore)
	if err != nil {
		logger.Errorf("Error parsing object store settings: %+v", err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	if objectStore.Gateway.Port == 0 && objectStore.Gateway.SecurePort == 0 {
		logger.Errorf("Must specify port or securePort")
		w.WriteHeader(http.StatusBadRequest)
	}

	// only the default store is supported through the rest api
	if objectStore.Name == "" {
		objectStore.Name = defaultObjectStoreName
	}
	if objectStore.Gateway.Instances == 0 {
		objectStore.Gateway.Instances = defaultRGWInstances
	}

	if err := h.config.ClusterHandler.EnableObjectStore(objectStore); err != nil {
		logger.Errorf("failed to create object store: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	logger.Debugf("created object store")
}

// RemoveObjectStore removes the object store from this cluster.
// DELETE
// /objectstore/{name}
func (h *Handler) RemoveObjectStore(w http.ResponseWriter, r *http.Request) {
	storeName := mux.Vars(r)["name"]

	if err := h.config.ClusterHandler.RemoveObjectStore(storeName); err != nil {
		logger.Errorf("failed to remove object store: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	logger.Debugf("started async deletion of the object store")
}

// GetObjectStoreConnectionInfo gets connection information to the object store in this cluster.
// GET
// /objectstore/{name}/connectioninfo
func (h *Handler) GetObjectStoreConnectionInfo(w http.ResponseWriter, r *http.Request) {
	storeName := mux.Vars(r)["name"]
	s3Info, found, err := h.config.ClusterHandler.GetObjectStoreConnectionInfo(storeName)
	if err != nil {
		logger.Errorf("failed get object store info. %+v", err)
		if found {
			w.WriteHeader(http.StatusInternalServerError)
		} else {
			w.WriteHeader(http.StatusNotFound)
		}
		return
	}

	FormatJsonResponse(w, s3Info)
}

// ListUsers lists the users of the object store in this cluster.
// GET
// /objectstore/{name}/users
func (h *Handler) ListUsers(w http.ResponseWriter, r *http.Request) {
	userNames, _, err := rgw.ListUsers(h.objectContext(r))
	if err != nil {
		logger.Errorf("Error listing users: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	users := []model.ObjectUser{}
	for _, userName := range userNames {
		user, _, err := rgw.GetUser(h.objectContext(r), userName)
		if err != nil {
			logger.Errorf("Error listing users: %+v", err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}

		users = append(users, *user)
	}

	FormatJsonResponse(w, users)
}

// GetUser gets the passed users info from the object store in this cluster.
// GET
// /objectstore/{name}/users/{USER_ID}
func (h *Handler) GetUser(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]

	user, rgwError, err := rgw.GetUser(h.objectContext(r), id)
	if err != nil {
		logger.Errorf("Error getting user (%s): %+v", id, err)

		if rgwError == rgw.RGWErrorNotFound {
			w.WriteHeader(http.StatusNotFound)
		} else {
			w.WriteHeader(http.StatusInternalServerError)
		}
		return
	}

	FormatJsonResponse(w, user)
}

// CreateUser will create a new user from the passed info in the object store in this cluster.
// POST
// /objectstore/{name}/users
func (h *Handler) CreateUser(w http.ResponseWriter, r *http.Request) {
	var user model.ObjectUser

	err := json.NewDecoder(r.Body).Decode(&user)
	if err != nil {
		logger.Errorf("Error parsing user: %+v", err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	createdUser, rgwError, err := rgw.CreateUser(h.objectContext(r), user)
	if err != nil {
		logger.Errorf("Error creating user: %+v", err)

		if rgwError == rgw.RGWErrorBadData {
			w.WriteHeader(http.StatusUnprocessableEntity)
			w.Write([]byte(err.Error()))
		} else {
			w.WriteHeader(http.StatusInternalServerError)
		}
		return
	}

	w.Header().Set("Content-Type", "application/json; charset=UTF-8")
	w.WriteHeader(http.StatusCreated)
	FormatJsonResponse(w, *createdUser)
}

// UpdateUser updates the passed user with the passed info for the object store in this cluster.
// PUT
// /objectstore/{name}/users/{USER_ID}
func (h *Handler) UpdateUser(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]

	var user model.ObjectUser
	err := json.NewDecoder(r.Body).Decode(&user)
	if err != nil {
		logger.Errorf("Error parsing user: %+v", err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	user.UserID = id

	updatedUser, rgwError, err := rgw.UpdateUser(h.objectContext(r), user)
	if err != nil {
		logger.Errorf("Error updating user: %+v", err)

		if rgwError == rgw.RGWErrorNotFound {
			w.WriteHeader(http.StatusNotFound)
		} else {
			w.WriteHeader(http.StatusInternalServerError)
		}
		return
	}

	FormatJsonResponse(w, updatedUser)
}

// DeleteUser deletes the passed user for the object store in this cluster.
// DELETE
// /objectstore/{name}/users/{USER_ID}
func (h *Handler) DeleteUser(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]

	_, rgwError, err := rgw.DeleteUser(h.objectContext(r), id)
	if err != nil {
		if rgwError == rgw.RGWErrorNotFound {
			w.WriteHeader(http.StatusNotFound)
			return
		}

		logger.Errorf("Error deleting user (%s): %+v", id, err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusNoContent)
}

// Listbuckets lists the buckets in the object store in this cluster.
// GET
// /objectstore/{name}/buckets
func (h *Handler) ListBuckets(w http.ResponseWriter, r *http.Request) {
	buckets, err := rgw.ListBuckets(h.objectContext(r))
	if err != nil {
		logger.Errorf("Error listing buckets: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	sortedBuckets := model.ObjectBuckets(buckets)
	sort.Sort(sortedBuckets)

	FormatJsonResponse(w, sortedBuckets)
}

// GetBucket gets the bucket from the object store in this cluster.
// GET
// /objectstore/{name}/buckets/{BUCKET_NAME}
func (h *Handler) GetBucket(w http.ResponseWriter, r *http.Request) {
	bucketName := mux.Vars(r)["bucketName"]

	user, rgwError, err := rgw.GetBucket(h.objectContext(r), bucketName)
	if err != nil {
		logger.Errorf("Error getting bucket (%s): %+v", bucketName, err)

		if rgwError == rgw.RGWErrorNotFound {
			w.WriteHeader(http.StatusNotFound)
		} else {
			w.WriteHeader(http.StatusInternalServerError)
		}
		return
	}

	FormatJsonResponse(w, user)
}

// DeleteBucket deletes the bucket in the object store in this cluster.
// DELETE
// /objectstore/{name}/buckets/{BUCKET_NAME}
func (h *Handler) DeleteBucket(w http.ResponseWriter, r *http.Request) {
	bucketName := mux.Vars(r)["bucketName"]

	purgeParams, found := r.URL.Query()["purge"]
	purge := found && len(purgeParams) == 1 && purgeParams[0] == "true"

	rgwError, err := rgw.DeleteBucket(h.objectContext(r), bucketName, purge)
	if err != nil {
		if rgwError == rgw.RGWErrorNotFound {
			w.WriteHeader(http.StatusNotFound)
			return
		}

		logger.Errorf("Error deleting bucket: %+v", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusNoContent)
}