feat(tenant): Add in api elements to tenant service (#17447)
Co-authored-by: Alirie Gray <alirie.gray@gmail.com> Co-authored-by: Alirie Gray <alirie.gray@gmail.com>pull/17465/head
Lyon Hill
GitHub
parent
ad38ed1215
commit
1a66ca3900
|
|
@ -8,6 +8,8 @@ import (
|
|||
"github.com/influxdata/influxdb"
|
||||
)
|
||||
|
||||
// TODO remove this file once bucket and org service are moved to the tenant service
|
||||
|
||||
func decodeIDFromCtx(ctx context.Context, name string) (influxdb.ID, error) {
|
||||
params := httprouter.ParamsFromContext(ctx)
|
||||
idStr := params.ByName(name)
|
||||
|
|
|
|||
|
|
@ -248,13 +248,13 @@ func (h *OrgHandler) handleGetOrgs(w http.ResponseWriter, r *http.Request) {
|
|||
if name := qp.Get(Org); name != "" {
|
||||
filter.Name = &name
|
||||
}
|
||||
id, err := decodeIDFromQuery(qp, OrgID)
|
||||
if err != nil {
|
||||
h.API.Err(w, err)
|
||||
return
|
||||
}
|
||||
if id > 0 {
|
||||
filter.ID = &id
|
||||
if orgID := qp.Get("orgID"); orgID != "" {
|
||||
id, err := influxdb.IDFromString(orgID)
|
||||
if err != nil {
|
||||
h.API.Err(w, err)
|
||||
return
|
||||
}
|
||||
filter.ID = id
|
||||
}
|
||||
|
||||
orgs, _, err := h.OrgSVC.FindOrganizations(r.Context(), filter)
|
||||
|
|
|
|||
55
paging.go
55
paging.go
|
|
@ -1,6 +1,7 @@
|
|||
package influxdb
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
|
|
@ -47,3 +48,57 @@ func (f FindOptions) QueryParams() map[string][]string {
|
|||
|
||||
return qp
|
||||
}
|
||||
|
||||
// NewPagingLinks returns a PagingLinks.
|
||||
// num is the number of returned results.
|
||||
func NewPagingLinks(basePath string, opts FindOptions, f PagingFilter, num int) *PagingLinks {
|
||||
u := url.URL{
|
||||
Path: basePath,
|
||||
}
|
||||
|
||||
values := url.Values{}
|
||||
for k, vs := range f.QueryParams() {
|
||||
for _, v := range vs {
|
||||
if v != "" {
|
||||
values.Add(k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var self, next, prev string
|
||||
for k, vs := range opts.QueryParams() {
|
||||
for _, v := range vs {
|
||||
if v != "" {
|
||||
values.Add(k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
u.RawQuery = values.Encode()
|
||||
self = u.String()
|
||||
|
||||
if num >= opts.Limit {
|
||||
nextOffset := opts.Offset + opts.Limit
|
||||
values.Set("offset", strconv.Itoa(nextOffset))
|
||||
u.RawQuery = values.Encode()
|
||||
next = u.String()
|
||||
}
|
||||
|
||||
if opts.Offset > 0 {
|
||||
prevOffset := opts.Offset - opts.Limit
|
||||
if prevOffset < 0 {
|
||||
prevOffset = 0
|
||||
}
|
||||
values.Set("offset", strconv.Itoa(prevOffset))
|
||||
u.RawQuery = values.Encode()
|
||||
prev = u.String()
|
||||
}
|
||||
|
||||
links := &PagingLinks{
|
||||
Prev: prev,
|
||||
Self: self,
|
||||
Next: next,
|
||||
}
|
||||
|
||||
return links
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,179 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"path"
|
||||
|
||||
"github.com/influxdata/influxdb"
|
||||
"github.com/influxdata/influxdb/kit/tracing"
|
||||
"github.com/influxdata/influxdb/pkg/httpc"
|
||||
)
|
||||
|
||||
// BucketClientService connects to Influx via HTTP using tokens to manage buckets
|
||||
type BucketClientService struct {
|
||||
Client *httpc.Client
|
||||
// OpPrefix is an additional property for error
|
||||
// find bucket service, when finds nothing.
|
||||
OpPrefix string
|
||||
}
|
||||
|
||||
// FindBucketByName returns a single bucket by name
|
||||
func (s *BucketClientService) FindBucketByName(ctx context.Context, orgID influxdb.ID, name string) (*influxdb.Bucket, error) {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
if name == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EUnprocessableEntity,
|
||||
Op: s.OpPrefix + influxdb.OpFindBuckets,
|
||||
Msg: "bucket name is required",
|
||||
}
|
||||
}
|
||||
|
||||
bkts, n, err := s.FindBuckets(ctx, influxdb.BucketFilter{
|
||||
Name: &name,
|
||||
OrganizationID: &orgID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if n == 0 || len(bkts) == 0 {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Op: s.OpPrefix + influxdb.OpFindBucket,
|
||||
Msg: fmt.Sprintf("bucket %q not found", name),
|
||||
}
|
||||
}
|
||||
|
||||
return bkts[0], nil
|
||||
}
|
||||
|
||||
// FindBucketByID returns a single bucket by ID.
|
||||
func (s *BucketClientService) FindBucketByID(ctx context.Context, id influxdb.ID) (*influxdb.Bucket, error) {
|
||||
// TODO(@jsteenb2): are tracing
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
var br bucketResponse
|
||||
err := s.Client.
|
||||
Get(path.Join(prefixBuckets, id.String())).
|
||||
DecodeJSON(&br).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return br.toInfluxDB()
|
||||
}
|
||||
|
||||
// FindBucket returns the first bucket that matches filter.
|
||||
func (s *BucketClientService) FindBucket(ctx context.Context, filter influxdb.BucketFilter) (*influxdb.Bucket, error) {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
bs, n, err := s.FindBuckets(ctx, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if n == 0 && filter.Name != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Op: s.OpPrefix + influxdb.OpFindBucket,
|
||||
Msg: fmt.Sprintf("bucket %q not found", *filter.Name),
|
||||
}
|
||||
} else if n == 0 {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Op: s.OpPrefix + influxdb.OpFindBucket,
|
||||
Msg: "bucket not found",
|
||||
}
|
||||
}
|
||||
|
||||
return bs[0], nil
|
||||
}
|
||||
|
||||
// FindBuckets returns a list of buckets that match filter and the total count of matching buckets.
|
||||
// Additional options provide pagination & sorting.
|
||||
func (s *BucketClientService) FindBuckets(ctx context.Context, filter influxdb.BucketFilter, opt ...influxdb.FindOptions) ([]*influxdb.Bucket, int, error) {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
params := findOptionParams(opt...)
|
||||
if filter.OrganizationID != nil {
|
||||
params = append(params, [2]string{"orgID", filter.OrganizationID.String()})
|
||||
}
|
||||
if filter.Org != nil {
|
||||
params = append(params, [2]string{"org", *filter.Org})
|
||||
}
|
||||
if filter.ID != nil {
|
||||
params = append(params, [2]string{"id", filter.ID.String()})
|
||||
}
|
||||
if filter.Name != nil {
|
||||
params = append(params, [2]string{"name", (*filter.Name)})
|
||||
}
|
||||
|
||||
var bs bucketsResponse
|
||||
err := s.Client.
|
||||
Get(prefixBuckets).
|
||||
QueryParams(params...).
|
||||
DecodeJSON(&bs).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
buckets := make([]*influxdb.Bucket, 0, len(bs.Buckets))
|
||||
for _, b := range bs.Buckets {
|
||||
pb, err := b.bucket.toInfluxDB()
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
buckets = append(buckets, pb)
|
||||
}
|
||||
|
||||
return buckets, len(buckets), nil
|
||||
}
|
||||
|
||||
// CreateBucket creates a new bucket and sets b.ID with the new identifier.
|
||||
func (s *BucketClientService) CreateBucket(ctx context.Context, b *influxdb.Bucket) error {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
var br bucketResponse
|
||||
err := s.Client.
|
||||
PostJSON(newBucket(b), prefixBuckets).
|
||||
DecodeJSON(&br).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pb, err := br.toInfluxDB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
*b = *pb
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateBucket updates a single bucket with changeset.
|
||||
// Returns the new bucket state after update.
|
||||
func (s *BucketClientService) UpdateBucket(ctx context.Context, id influxdb.ID, upd influxdb.BucketUpdate) (*influxdb.Bucket, error) {
|
||||
var br bucketResponse
|
||||
err := s.Client.
|
||||
PatchJSON(newBucketUpdate(&upd), path.Join(prefixBuckets, id.String())).
|
||||
DecodeJSON(&br).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return br.toInfluxDB()
|
||||
}
|
||||
|
||||
// DeleteBucket removes a bucket by ID.
|
||||
func (s *BucketClientService) DeleteBucket(ctx context.Context, id influxdb.ID) error {
|
||||
return s.Client.
|
||||
Delete(path.Join(prefixBuckets, id.String())).
|
||||
Do(ctx)
|
||||
}
|
||||
|
|
@ -0,0 +1,150 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/influxdata/influxdb"
|
||||
"github.com/influxdata/influxdb/kit/tracing"
|
||||
"github.com/influxdata/influxdb/pkg/httpc"
|
||||
)
|
||||
|
||||
// OrgClientService connects to Influx via HTTP using tokens to manage organizations
|
||||
type OrgClientService struct {
|
||||
Client *httpc.Client
|
||||
// OpPrefix is for not found errors.
|
||||
OpPrefix string
|
||||
}
|
||||
|
||||
func (o orgsResponse) toInfluxdb() []*influxdb.Organization {
|
||||
orgs := make([]*influxdb.Organization, len(o.Organizations))
|
||||
for i := range o.Organizations {
|
||||
orgs[i] = &o.Organizations[i].Organization
|
||||
}
|
||||
return orgs
|
||||
}
|
||||
|
||||
// FindOrganizationByID gets a single organization with a given id using HTTP.
|
||||
func (s *OrgClientService) FindOrganizationByID(ctx context.Context, id influxdb.ID) (*influxdb.Organization, error) {
|
||||
filter := influxdb.OrganizationFilter{ID: &id}
|
||||
o, err := s.FindOrganization(ctx, filter)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Err: err,
|
||||
Op: s.OpPrefix + influxdb.OpFindOrganizationByID,
|
||||
}
|
||||
}
|
||||
return o, nil
|
||||
}
|
||||
|
||||
// FindOrganization gets a single organization matching the filter using HTTP.
|
||||
func (s *OrgClientService) FindOrganization(ctx context.Context, filter influxdb.OrganizationFilter) (*influxdb.Organization, error) {
|
||||
if filter.ID == nil && filter.Name == nil {
|
||||
return nil, influxdb.ErrInvalidOrgFilter
|
||||
}
|
||||
os, n, err := s.FindOrganizations(ctx, filter)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Err: err,
|
||||
Op: s.OpPrefix + influxdb.OpFindOrganization,
|
||||
}
|
||||
}
|
||||
|
||||
if n == 0 {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Op: s.OpPrefix + influxdb.OpFindOrganization,
|
||||
Msg: "organization not found",
|
||||
}
|
||||
}
|
||||
|
||||
return os[0], nil
|
||||
}
|
||||
|
||||
// FindOrganizations returns all organizations that match the filter via HTTP.
|
||||
func (s *OrgClientService) FindOrganizations(ctx context.Context, filter influxdb.OrganizationFilter, opt ...influxdb.FindOptions) ([]*influxdb.Organization, int, error) {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
params := findOptionParams(opt...)
|
||||
if filter.Name != nil {
|
||||
span.LogKV("org", *filter.Name)
|
||||
params = append(params, [2]string{"org", *filter.Name})
|
||||
}
|
||||
if filter.ID != nil {
|
||||
span.LogKV("org-id", *filter.ID)
|
||||
params = append(params, [2]string{"orgID", filter.ID.String()})
|
||||
}
|
||||
for _, o := range opt {
|
||||
if o.Offset != 0 {
|
||||
span.LogKV("offset", o.Offset)
|
||||
}
|
||||
span.LogKV("descending", o.Descending)
|
||||
if o.Limit > 0 {
|
||||
span.LogKV("limit", o.Limit)
|
||||
}
|
||||
if o.SortBy != "" {
|
||||
span.LogKV("sortBy", o.SortBy)
|
||||
}
|
||||
}
|
||||
|
||||
var os orgsResponse
|
||||
err := s.Client.
|
||||
Get(prefixOrganizations).
|
||||
QueryParams(params...).
|
||||
DecodeJSON(&os).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
orgs := os.toInfluxdb()
|
||||
return orgs, len(orgs), nil
|
||||
}
|
||||
|
||||
// CreateOrganization creates an organization.
|
||||
func (s *OrgClientService) CreateOrganization(ctx context.Context, o *influxdb.Organization) error {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
if o.Name != "" {
|
||||
span.LogKV("org", o.Name)
|
||||
}
|
||||
if o.ID != 0 {
|
||||
span.LogKV("org-id", o.ID)
|
||||
}
|
||||
|
||||
return s.Client.
|
||||
PostJSON(o, prefixOrganizations).
|
||||
DecodeJSON(o).
|
||||
Do(ctx)
|
||||
}
|
||||
|
||||
// UpdateOrganization updates the organization over HTTP.
|
||||
func (s *OrgClientService) UpdateOrganization(ctx context.Context, id influxdb.ID, upd influxdb.OrganizationUpdate) (*influxdb.Organization, error) {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
span.LogKV("org-id", id)
|
||||
span.LogKV("name", upd.Name)
|
||||
|
||||
var o influxdb.Organization
|
||||
err := s.Client.
|
||||
PatchJSON(upd, prefixOrganizations, id.String()).
|
||||
DecodeJSON(&o).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, tracing.LogError(span, err)
|
||||
}
|
||||
|
||||
return &o, nil
|
||||
}
|
||||
|
||||
// DeleteOrganization removes organization id over HTTP.
|
||||
func (s *OrgClientService) DeleteOrganization(ctx context.Context, id influxdb.ID) error {
|
||||
span, _ := tracing.StartSpanFromContext(ctx)
|
||||
defer span.Finish()
|
||||
|
||||
return s.Client.
|
||||
Delete(prefixOrganizations, id.String()).
|
||||
Do(ctx)
|
||||
}
|
||||
|
|
@ -0,0 +1,154 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/influxdata/influxdb"
|
||||
ihttp "github.com/influxdata/influxdb/http"
|
||||
"github.com/influxdata/influxdb/pkg/httpc"
|
||||
)
|
||||
|
||||
// UserService connects to Influx via HTTP using tokens to manage users
|
||||
type UserClientService struct {
|
||||
Client *httpc.Client
|
||||
// OpPrefix is the ops of not found error.
|
||||
OpPrefix string
|
||||
}
|
||||
|
||||
// FindMe returns user information about the owner of the token
|
||||
func (s *UserClientService) FindMe(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
var res userResponse
|
||||
err := s.Client.
|
||||
Get(prefixMe).
|
||||
DecodeJSON(&res).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &res.User, nil
|
||||
}
|
||||
|
||||
// FindUserByID returns a single user by ID.
|
||||
func (s *UserClientService) FindUserByID(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
var res userResponse
|
||||
err := s.Client.
|
||||
Get(prefixUsers, id.String()).
|
||||
DecodeJSON(&res).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &res.User, nil
|
||||
}
|
||||
|
||||
// FindUser returns the first user that matches filter.
|
||||
func (s *UserClientService) FindUser(ctx context.Context, filter influxdb.UserFilter) (*influxdb.User, error) {
|
||||
if filter.ID == nil && filter.Name == nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Msg: "user not found",
|
||||
}
|
||||
}
|
||||
users, n, err := s.FindUsers(ctx, filter)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Op: s.OpPrefix + influxdb.OpFindUser,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
if n == 0 {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.ENotFound,
|
||||
Op: s.OpPrefix + influxdb.OpFindUser,
|
||||
Msg: "no results found",
|
||||
}
|
||||
}
|
||||
|
||||
return users[0], nil
|
||||
}
|
||||
|
||||
// FindUsers returns a list of users that match filter and the total count of matching users.
|
||||
// Additional options provide pagination & sorting.
|
||||
func (s *UserClientService) FindUsers(ctx context.Context, filter influxdb.UserFilter, opt ...influxdb.FindOptions) ([]*influxdb.User, int, error) {
|
||||
params := findOptionParams(opt...)
|
||||
if filter.ID != nil {
|
||||
params = append(params, [2]string{"id", filter.ID.String()})
|
||||
}
|
||||
if filter.Name != nil {
|
||||
params = append(params, [2]string{"name", *filter.Name})
|
||||
}
|
||||
|
||||
var r usersResponse
|
||||
err := s.Client.
|
||||
Get(prefixUsers).
|
||||
QueryParams(params...).
|
||||
DecodeJSON(&r).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
us := r.ToInfluxdb()
|
||||
return us, len(us), nil
|
||||
}
|
||||
|
||||
// CreateUser creates a new user and sets u.ID with the new identifier.
|
||||
func (s *UserClientService) CreateUser(ctx context.Context, u *influxdb.User) error {
|
||||
return s.Client.
|
||||
PostJSON(u, prefixUsers).
|
||||
DecodeJSON(u).
|
||||
Do(ctx)
|
||||
}
|
||||
|
||||
// UpdateUser updates a single user with changeset.
|
||||
// Returns the new user state after update.
|
||||
func (s *UserClientService) UpdateUser(ctx context.Context, id influxdb.ID, upd influxdb.UserUpdate) (*influxdb.User, error) {
|
||||
var res userResponse
|
||||
err := s.Client.
|
||||
PatchJSON(upd, prefixUsers, id.String()).
|
||||
DecodeJSON(&res).
|
||||
Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &res.User, nil
|
||||
}
|
||||
|
||||
// DeleteUser removes a user by ID.
|
||||
func (s *UserClientService) DeleteUser(ctx context.Context, id influxdb.ID) error {
|
||||
return s.Client.
|
||||
Delete(prefixUsers, id.String()).
|
||||
StatusFn(func(resp *http.Response) error {
|
||||
return ihttp.CheckErrorStatus(http.StatusNoContent, resp)
|
||||
}).
|
||||
Do(ctx)
|
||||
}
|
||||
|
||||
// PasswordClientService is an http client to speak to the password service.
|
||||
type PasswordClientService struct {
|
||||
Client *httpc.Client
|
||||
}
|
||||
|
||||
var _ influxdb.PasswordsService = (*PasswordClientService)(nil)
|
||||
|
||||
// SetPassword sets the user's password.
|
||||
func (s *PasswordClientService) SetPassword(ctx context.Context, userID influxdb.ID, password string) error {
|
||||
return s.Client.
|
||||
PostJSON(passwordSetRequest{
|
||||
Password: password,
|
||||
}, prefixUsers, userID.String(), "password").
|
||||
Do(ctx)
|
||||
}
|
||||
|
||||
// ComparePassword compares the user new password with existing. Note: is not implemented.
|
||||
func (s *PasswordClientService) ComparePassword(ctx context.Context, userID influxdb.ID, password string) error {
|
||||
panic("not implemented")
|
||||
}
|
||||
|
||||
// CompareAndSetPassword compares the old and new password and submits the new password if possoble.
|
||||
// Note: is not implemented.
|
||||
func (s *PasswordClientService) CompareAndSetPassword(ctx context.Context, userID influxdb.ID, old string, new string) error {
|
||||
panic("not implemented")
|
||||
}
|
||||
|
|
@ -0,0 +1,277 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"path"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/influxdata/influxdb"
|
||||
kithttp "github.com/influxdata/influxdb/kit/transport/http"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type urmHandler struct {
|
||||
log *zap.Logger
|
||||
svc influxdb.UserResourceMappingService
|
||||
userSvc influxdb.UserService
|
||||
api *kithttp.API
|
||||
|
||||
rt influxdb.ResourceType
|
||||
idLookupKey string
|
||||
}
|
||||
|
||||
// NewURMHandler generates a mountable handler for URMs. It needs to know how it will be looking up your resource id
|
||||
// this system assumes you are using chi syntax for query string params `/orgs/{id}/` so it can use chi.URLParam().
|
||||
func NewURMHandler(log *zap.Logger, rt influxdb.ResourceType, idLookupKey string, uSvc influxdb.UserService, urmSvc influxdb.UserResourceMappingService) http.Handler {
|
||||
h := &urmHandler{
|
||||
log: log,
|
||||
svc: urmSvc,
|
||||
userSvc: uSvc,
|
||||
api: kithttp.NewAPI(kithttp.WithLog(log)),
|
||||
|
||||
rt: rt,
|
||||
idLookupKey: idLookupKey,
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Get("/", h.getURMsByType)
|
||||
r.Post("/", h.postURMByType)
|
||||
r.Delete("/{userID}", h.deleteURM)
|
||||
return r
|
||||
}
|
||||
|
||||
func (h *urmHandler) getURMsByType(w http.ResponseWriter, r *http.Request) {
|
||||
userType := userTypeFromPath(r.URL.Path)
|
||||
ctx := r.Context()
|
||||
req, err := h.decodeGetRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
filter := influxdb.UserResourceMappingFilter{
|
||||
ResourceID: req.ResourceID,
|
||||
ResourceType: h.rt,
|
||||
UserType: userType,
|
||||
}
|
||||
mappings, _, err := h.svc.FindUserResourceMappings(ctx, filter)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
users := make([]*influxdb.User, 0, len(mappings))
|
||||
for _, m := range mappings {
|
||||
if m.MappingType == influxdb.OrgMappingType {
|
||||
continue
|
||||
}
|
||||
user, err := h.userSvc.FindUserByID(ctx, m.UserID)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
users = append(users, user)
|
||||
}
|
||||
h.log.Debug("Members/owners retrieved", zap.String("users", fmt.Sprint(users)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newResourceUsersResponse(filter, users))
|
||||
|
||||
}
|
||||
|
||||
type getRequest struct {
|
||||
ResourceID influxdb.ID
|
||||
}
|
||||
|
||||
func (h *urmHandler) decodeGetRequest(ctx context.Context, r *http.Request) (*getRequest, error) {
|
||||
id := chi.URLParam(r, h.idLookupKey)
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var i influxdb.ID
|
||||
if err := i.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req := &getRequest{
|
||||
ResourceID: i,
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func (h *urmHandler) postURMByType(w http.ResponseWriter, r *http.Request) {
|
||||
userType := userTypeFromPath(r.URL.Path)
|
||||
ctx := r.Context()
|
||||
req, err := h.decodePostRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
user, err := h.userSvc.FindUserByID(ctx, req.UserID)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
mapping := &influxdb.UserResourceMapping{
|
||||
ResourceID: req.ResourceID,
|
||||
ResourceType: h.rt,
|
||||
UserID: req.UserID,
|
||||
UserType: userType,
|
||||
}
|
||||
if err := h.svc.CreateUserResourceMapping(ctx, mapping); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Member/owner created", zap.String("mapping", fmt.Sprint(mapping)))
|
||||
|
||||
h.api.Respond(w, http.StatusCreated, newResourceUserResponse(user, userType))
|
||||
}
|
||||
|
||||
type postRequest struct {
|
||||
UserID influxdb.ID
|
||||
ResourceID influxdb.ID
|
||||
}
|
||||
|
||||
func (h urmHandler) decodePostRequest(ctx context.Context, r *http.Request) (*postRequest, error) {
|
||||
id := chi.URLParam(r, h.idLookupKey)
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var rid influxdb.ID
|
||||
if err := rid.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
u := &influxdb.User{}
|
||||
if err := json.NewDecoder(r.Body).Decode(u); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if !u.ID.Valid() {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "user id missing or invalid",
|
||||
}
|
||||
}
|
||||
|
||||
return &postRequest{
|
||||
UserID: u.ID,
|
||||
ResourceID: rid,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (h *urmHandler) deleteURM(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
req, err := h.decodeDeleteRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.svc.DeleteUserResourceMapping(ctx, req.resourceID, req.userID); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Member deleted", zap.String("resourceID", req.resourceID.String()), zap.String("memberID", req.userID.String()))
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
type deleteRequest struct {
|
||||
userID influxdb.ID
|
||||
resourceID influxdb.ID
|
||||
}
|
||||
|
||||
func (h *urmHandler) decodeDeleteRequest(ctx context.Context, r *http.Request) (*deleteRequest, error) {
|
||||
id := chi.URLParam(r, h.idLookupKey)
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var rid influxdb.ID
|
||||
if err := rid.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
id = chi.URLParam(r, "userID")
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing member id",
|
||||
}
|
||||
}
|
||||
|
||||
var uid influxdb.ID
|
||||
if err := uid.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &deleteRequest{
|
||||
userID: uid,
|
||||
resourceID: rid,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type resourceUserResponse struct {
|
||||
Role influxdb.UserType `json:"role"`
|
||||
*userResponse
|
||||
}
|
||||
|
||||
func newResourceUserResponse(u *influxdb.User, userType influxdb.UserType) *resourceUserResponse {
|
||||
return &resourceUserResponse{
|
||||
Role: userType,
|
||||
userResponse: newUserResponse(u),
|
||||
}
|
||||
}
|
||||
|
||||
type resourceUsersResponse struct {
|
||||
Links map[string]string `json:"links"`
|
||||
Users []*resourceUserResponse `json:"users"`
|
||||
}
|
||||
|
||||
func newResourceUsersResponse(f influxdb.UserResourceMappingFilter, users []*influxdb.User) *resourceUsersResponse {
|
||||
rs := resourceUsersResponse{
|
||||
Links: map[string]string{
|
||||
"self": fmt.Sprintf("/api/v2/%s/%s/%ss", f.ResourceType, f.ResourceID, f.UserType),
|
||||
},
|
||||
Users: make([]*resourceUserResponse, 0, len(users)),
|
||||
}
|
||||
|
||||
for _, user := range users {
|
||||
rs.Users = append(rs.Users, newResourceUserResponse(user, f.UserType))
|
||||
}
|
||||
return &rs
|
||||
}
|
||||
|
||||
// determine the type of request from the path.
|
||||
func userTypeFromPath(p string) influxdb.UserType {
|
||||
if p == "" {
|
||||
return influxdb.Member
|
||||
}
|
||||
|
||||
switch path.Base(p) {
|
||||
case "members":
|
||||
return influxdb.Member
|
||||
case "owners":
|
||||
return influxdb.Owner
|
||||
default:
|
||||
return userTypeFromPath(path.Dir(p))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,375 @@
|
|||
package tenant_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/influxdata/influxdb"
|
||||
"github.com/influxdata/influxdb/mock"
|
||||
"github.com/influxdata/influxdb/tenant"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
func TestUserResourceMappingService_GetMembersHandler(t *testing.T) {
|
||||
type fields struct {
|
||||
userService influxdb.UserService
|
||||
userResourceMappingService influxdb.UserResourceMappingService
|
||||
}
|
||||
type args struct {
|
||||
resourceID string
|
||||
userType influxdb.UserType
|
||||
}
|
||||
type wants struct {
|
||||
statusCode int
|
||||
contentType string
|
||||
body string
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
wants wants
|
||||
}{
|
||||
{
|
||||
name: "get members",
|
||||
fields: fields{
|
||||
userService: &mock.UserService{
|
||||
FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
return &influxdb.User{ID: id, Name: fmt.Sprintf("user%s", id), Status: influxdb.Active}, nil
|
||||
},
|
||||
},
|
||||
userResourceMappingService: &mock.UserResourceMappingService{
|
||||
FindMappingsFn: func(ctx context.Context, filter influxdb.UserResourceMappingFilter) ([]*influxdb.UserResourceMapping, int, error) {
|
||||
ms := []*influxdb.UserResourceMapping{
|
||||
{
|
||||
ResourceID: filter.ResourceID,
|
||||
ResourceType: filter.ResourceType,
|
||||
UserType: filter.UserType,
|
||||
UserID: 1,
|
||||
},
|
||||
{
|
||||
ResourceID: filter.ResourceID,
|
||||
ResourceType: filter.ResourceType,
|
||||
UserType: filter.UserType,
|
||||
UserID: 2,
|
||||
},
|
||||
}
|
||||
return ms, len(ms), nil
|
||||
},
|
||||
},
|
||||
},
|
||||
args: args{
|
||||
resourceID: "0000000000000099",
|
||||
userType: influxdb.Member,
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
body: `{
|
||||
"links": {
|
||||
"self": "/api/v2/%s/0000000000000099/members"
|
||||
},
|
||||
"users": [
|
||||
{
|
||||
"role": "member",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000001"
|
||||
},
|
||||
"id": "0000000000000001",
|
||||
"name": "user0000000000000001",
|
||||
"status": "active"
|
||||
},
|
||||
{
|
||||
"role": "member",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000002"
|
||||
},
|
||||
"id": "0000000000000002",
|
||||
"name": "user0000000000000002",
|
||||
"status": "active"
|
||||
}
|
||||
]
|
||||
}
|
||||
`,
|
||||
},
|
||||
},
|
||||
|
||||
{
|
||||
name: "get owners",
|
||||
fields: fields{
|
||||
userService: &mock.UserService{
|
||||
FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
return &influxdb.User{ID: id, Name: fmt.Sprintf("user%s", id), Status: influxdb.Active}, nil
|
||||
},
|
||||
},
|
||||
userResourceMappingService: &mock.UserResourceMappingService{
|
||||
FindMappingsFn: func(ctx context.Context, filter influxdb.UserResourceMappingFilter) ([]*influxdb.UserResourceMapping, int, error) {
|
||||
ms := []*influxdb.UserResourceMapping{
|
||||
{
|
||||
ResourceID: filter.ResourceID,
|
||||
ResourceType: filter.ResourceType,
|
||||
UserType: filter.UserType,
|
||||
UserID: 1,
|
||||
},
|
||||
{
|
||||
ResourceID: filter.ResourceID,
|
||||
ResourceType: filter.ResourceType,
|
||||
UserType: filter.UserType,
|
||||
UserID: 2,
|
||||
},
|
||||
}
|
||||
return ms, len(ms), nil
|
||||
},
|
||||
},
|
||||
},
|
||||
args: args{
|
||||
resourceID: "0000000000000099",
|
||||
userType: influxdb.Owner,
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusOK,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
body: `{
|
||||
"links": {
|
||||
"self": "/api/v2/%s/0000000000000099/owners"
|
||||
},
|
||||
"users": [
|
||||
{
|
||||
"role": "owner",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000001"
|
||||
},
|
||||
"id": "0000000000000001",
|
||||
"name": "user0000000000000001",
|
||||
"status": "active"
|
||||
},
|
||||
{
|
||||
"role": "owner",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000002"
|
||||
},
|
||||
"id": "0000000000000002",
|
||||
"name": "user0000000000000002",
|
||||
"status": "active"
|
||||
}
|
||||
]
|
||||
}
|
||||
`,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
resourceTypes := []influxdb.ResourceType{
|
||||
influxdb.BucketsResourceType,
|
||||
influxdb.DashboardsResourceType,
|
||||
influxdb.OrgsResourceType,
|
||||
influxdb.SourcesResourceType,
|
||||
influxdb.TasksResourceType,
|
||||
influxdb.TelegrafsResourceType,
|
||||
influxdb.UsersResourceType,
|
||||
}
|
||||
|
||||
for _, resourceType := range resourceTypes {
|
||||
t.Run(tt.name+"_"+string(resourceType), func(t *testing.T) {
|
||||
// create server
|
||||
h := tenant.NewURMHandler(zaptest.NewLogger(t), resourceType, "id", tt.fields.userService, tt.fields.userResourceMappingService)
|
||||
router := chi.NewRouter()
|
||||
router.Mount(fmt.Sprintf("/api/v2/%s/{id}/members", resourceType), h)
|
||||
router.Mount(fmt.Sprintf("/api/v2/%s/{id}/owners", resourceType), h)
|
||||
s := httptest.NewServer(router)
|
||||
defer s.Close()
|
||||
|
||||
// craft request
|
||||
r, err := http.NewRequest("GET", fmt.Sprintf("%s/api/v2/%s/%s/%ss", s.URL, resourceType, tt.args.resourceID, tt.args.userType), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
c := s.Client()
|
||||
res, err := c.Do(r)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// check response
|
||||
content := res.Header.Get("Content-Type")
|
||||
body, _ := ioutil.ReadAll(res.Body)
|
||||
if res.StatusCode != tt.wants.statusCode {
|
||||
t.Errorf("%q. GetMembersHandler() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
|
||||
}
|
||||
if tt.wants.contentType != "" && content != tt.wants.contentType {
|
||||
t.Errorf("%q. GetMembersHandler() = %v, want %v", tt.name, content, tt.wants.contentType)
|
||||
}
|
||||
if diff := cmp.Diff(string(body), fmt.Sprintf(tt.wants.body, resourceType)); tt.wants.body != "" && diff != "" {
|
||||
t.Errorf("%q. GetMembersHandler() = ***%s***", tt.name, diff)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserResourceMappingService_PostMembersHandler(t *testing.T) {
|
||||
type fields struct {
|
||||
userService influxdb.UserService
|
||||
userResourceMappingService influxdb.UserResourceMappingService
|
||||
}
|
||||
type args struct {
|
||||
resourceID string
|
||||
userType influxdb.UserType
|
||||
user influxdb.User
|
||||
}
|
||||
type wants struct {
|
||||
statusCode int
|
||||
contentType string
|
||||
body string
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
wants wants
|
||||
}{
|
||||
{
|
||||
name: "post members",
|
||||
fields: fields{
|
||||
userService: &mock.UserService{
|
||||
FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
return &influxdb.User{ID: id, Name: fmt.Sprintf("user%s", id), Status: influxdb.Active}, nil
|
||||
},
|
||||
},
|
||||
userResourceMappingService: &mock.UserResourceMappingService{
|
||||
CreateMappingFn: func(ctx context.Context, m *influxdb.UserResourceMapping) error {
|
||||
return nil
|
||||
},
|
||||
},
|
||||
},
|
||||
args: args{
|
||||
resourceID: "0000000000000099",
|
||||
user: influxdb.User{
|
||||
ID: 1,
|
||||
Name: "user0000000000000001",
|
||||
Status: influxdb.Active,
|
||||
},
|
||||
userType: influxdb.Member,
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusCreated,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
body: `{
|
||||
"role": "member",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000001"
|
||||
},
|
||||
"id": "0000000000000001",
|
||||
"name": "user0000000000000001",
|
||||
"status": "active"
|
||||
}
|
||||
`,
|
||||
},
|
||||
},
|
||||
|
||||
{
|
||||
name: "post owners",
|
||||
fields: fields{
|
||||
userService: &mock.UserService{
|
||||
FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
|
||||
return &influxdb.User{ID: id, Name: fmt.Sprintf("user%s", id), Status: influxdb.Active}, nil
|
||||
},
|
||||
},
|
||||
userResourceMappingService: &mock.UserResourceMappingService{
|
||||
CreateMappingFn: func(ctx context.Context, m *influxdb.UserResourceMapping) error {
|
||||
return nil
|
||||
},
|
||||
},
|
||||
},
|
||||
args: args{
|
||||
resourceID: "0000000000000099",
|
||||
user: influxdb.User{
|
||||
ID: 2,
|
||||
Name: "user0000000000000002",
|
||||
Status: influxdb.Active,
|
||||
},
|
||||
userType: influxdb.Owner,
|
||||
},
|
||||
wants: wants{
|
||||
statusCode: http.StatusCreated,
|
||||
contentType: "application/json; charset=utf-8",
|
||||
body: `{
|
||||
"role": "owner",
|
||||
"links": {
|
||||
"self": "/api/v2/users/0000000000000002"
|
||||
},
|
||||
"id": "0000000000000002",
|
||||
"name": "user0000000000000002",
|
||||
"status": "active"
|
||||
}
|
||||
`,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
resourceTypes := []influxdb.ResourceType{
|
||||
influxdb.BucketsResourceType,
|
||||
influxdb.DashboardsResourceType,
|
||||
influxdb.OrgsResourceType,
|
||||
influxdb.SourcesResourceType,
|
||||
influxdb.TasksResourceType,
|
||||
influxdb.TelegrafsResourceType,
|
||||
influxdb.UsersResourceType,
|
||||
}
|
||||
|
||||
for _, resourceType := range resourceTypes {
|
||||
t.Run(tt.name+"_"+string(resourceType), func(t *testing.T) {
|
||||
// create server
|
||||
h := tenant.NewURMHandler(zaptest.NewLogger(t), resourceType, "id", tt.fields.userService, tt.fields.userResourceMappingService)
|
||||
router := chi.NewRouter()
|
||||
router.Mount(fmt.Sprintf("/api/v2/%s/{id}/members", resourceType), h)
|
||||
router.Mount(fmt.Sprintf("/api/v2/%s/{id}/owners", resourceType), h)
|
||||
s := httptest.NewServer(router)
|
||||
defer s.Close()
|
||||
|
||||
// craft request
|
||||
b, err := json.Marshal(tt.args.user)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to unmarshal user: %v", err)
|
||||
}
|
||||
|
||||
r, err := http.NewRequest("POST", fmt.Sprintf("%s/api/v2/%s/%s/%ss", s.URL, resourceType, tt.args.resourceID, tt.args.userType), bytes.NewReader(b))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
c := s.Client()
|
||||
res, err := c.Do(r)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
content := res.Header.Get("Content-Type")
|
||||
body, _ := ioutil.ReadAll(res.Body)
|
||||
|
||||
if res.StatusCode != tt.wants.statusCode {
|
||||
t.Errorf("%q. PostMembersHandler() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
|
||||
}
|
||||
if tt.wants.contentType != "" && content != tt.wants.contentType {
|
||||
t.Errorf("%q. PostMembersHandler() = %v, want %v", tt.name, content, tt.wants.contentType)
|
||||
}
|
||||
if tt.wants.body != "" {
|
||||
if diff := cmp.Diff(string(body), tt.wants.body); diff != "" {
|
||||
t.Errorf("%q. PostMembersHandler() = ***%s***", tt.name, diff)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/influxdata/influxdb"
|
||||
)
|
||||
|
||||
// findOptionsParams converts find options into a paramiterizated key pair
|
||||
func findOptionParams(opts ...influxdb.FindOptions) [][2]string {
|
||||
var out [][2]string
|
||||
for _, o := range opts {
|
||||
for k, vals := range o.QueryParams() {
|
||||
for _, v := range vals {
|
||||
out = append(out, [2]string{k, v})
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// decodeFindOptions returns a FindOptions decoded from http request.
|
||||
func decodeFindOptions(r *http.Request) (*influxdb.FindOptions, error) {
|
||||
opts := &influxdb.FindOptions{}
|
||||
qp := r.URL.Query()
|
||||
|
||||
if offset := qp.Get("offset"); offset != "" {
|
||||
o, err := strconv.Atoi(offset)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "offset is invalid",
|
||||
}
|
||||
}
|
||||
|
||||
opts.Offset = o
|
||||
}
|
||||
|
||||
if limit := qp.Get("limit"); limit != "" {
|
||||
l, err := strconv.Atoi(limit)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "limit is invalid",
|
||||
}
|
||||
}
|
||||
|
||||
if l < 1 || l > influxdb.MaxPageSize {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: fmt.Sprintf("limit must be between 1 and %d", influxdb.MaxPageSize),
|
||||
}
|
||||
}
|
||||
|
||||
opts.Limit = l
|
||||
} else {
|
||||
opts.Limit = influxdb.DefaultPageSize
|
||||
}
|
||||
|
||||
if sortBy := qp.Get("sortBy"); sortBy != "" {
|
||||
opts.SortBy = sortBy
|
||||
}
|
||||
|
||||
if descending := qp.Get("descending"); descending != "" {
|
||||
desc, err := strconv.ParseBool(descending)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "descending is invalid",
|
||||
}
|
||||
}
|
||||
|
||||
opts.Descending = desc
|
||||
}
|
||||
|
||||
return opts, nil
|
||||
}
|
||||
|
|
@ -0,0 +1,479 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/go-chi/chi/middleware"
|
||||
"github.com/influxdata/influxdb"
|
||||
kithttp "github.com/influxdata/influxdb/kit/transport/http"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// BucketHandler represents an HTTP API handler for users.
|
||||
type BucketHandler struct {
|
||||
chi.Router
|
||||
api *kithttp.API
|
||||
log *zap.Logger
|
||||
bucketSvc influxdb.BucketService
|
||||
}
|
||||
|
||||
const (
|
||||
prefixBuckets = "/api/v2/buckets"
|
||||
)
|
||||
|
||||
// NewHTTPBucketHandler constructs a new http server.
|
||||
func NewHTTPBucketHandler(log *zap.Logger, bucketSvc influxdb.BucketService, urmHandler, labelHandler http.Handler) *BucketHandler {
|
||||
svr := &BucketHandler{
|
||||
api: kithttp.NewAPI(kithttp.WithLog(log)),
|
||||
log: log,
|
||||
bucketSvc: bucketSvc,
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Use(
|
||||
middleware.Recoverer,
|
||||
middleware.RequestID,
|
||||
middleware.RealIP,
|
||||
)
|
||||
|
||||
// RESTy routes for "articles" resource
|
||||
r.Route("/", func(r chi.Router) {
|
||||
r.Post("/", svr.handlePostBucket)
|
||||
r.Get("/", svr.handleGetBuckets)
|
||||
|
||||
r.Route("/{id}", func(r chi.Router) {
|
||||
r.Get("/", svr.handleGetBucket)
|
||||
r.Patch("/", svr.handlePatchBucket)
|
||||
r.Delete("/", svr.handleDeleteBucket)
|
||||
|
||||
// mount embedded resources
|
||||
r.Mount("/members", urmHandler)
|
||||
r.Mount("/owners", urmHandler)
|
||||
r.Mount("/labels", labelHandler)
|
||||
})
|
||||
})
|
||||
|
||||
svr.Router = r
|
||||
return svr
|
||||
}
|
||||
|
||||
func (h *BucketHandler) Prefix() string {
|
||||
return prefixBuckets
|
||||
}
|
||||
|
||||
// bucket is used for serialization/deserialization with duration string syntax.
|
||||
type bucket struct {
|
||||
ID influxdb.ID `json:"id,omitempty"`
|
||||
OrgID influxdb.ID `json:"orgID,omitempty"`
|
||||
Type string `json:"type"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Name string `json:"name"`
|
||||
RetentionPolicyName string `json:"rp,omitempty"` // This to support v1 sources
|
||||
RetentionRules []retentionRule `json:"retentionRules"`
|
||||
influxdb.CRUDLog
|
||||
}
|
||||
|
||||
// retentionRule is the retention rule action for a bucket.
|
||||
type retentionRule struct {
|
||||
Type string `json:"type"`
|
||||
EverySeconds int64 `json:"everySeconds"`
|
||||
}
|
||||
|
||||
func (rr *retentionRule) RetentionPeriod() (time.Duration, error) {
|
||||
t := time.Duration(rr.EverySeconds) * time.Second
|
||||
if t < time.Second {
|
||||
return t, &influxdb.Error{
|
||||
Code: influxdb.EUnprocessableEntity,
|
||||
Msg: "expiration seconds must be greater than or equal to one second",
|
||||
}
|
||||
}
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
func (b *bucket) toInfluxDB() (*influxdb.Bucket, error) {
|
||||
if b == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var d time.Duration // zero value implies infinite retention policy
|
||||
|
||||
// Only support a single retention period for the moment
|
||||
if len(b.RetentionRules) > 0 {
|
||||
d = time.Duration(b.RetentionRules[0].EverySeconds) * time.Second
|
||||
if d < time.Second {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EUnprocessableEntity,
|
||||
Msg: "expiration seconds must be greater than or equal to one second",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return &influxdb.Bucket{
|
||||
ID: b.ID,
|
||||
OrgID: b.OrgID,
|
||||
Type: influxdb.ParseBucketType(b.Type),
|
||||
Description: b.Description,
|
||||
Name: b.Name,
|
||||
RetentionPolicyName: b.RetentionPolicyName,
|
||||
RetentionPeriod: d,
|
||||
CRUDLog: b.CRUDLog,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func newBucket(pb *influxdb.Bucket) *bucket {
|
||||
if pb == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
rules := []retentionRule{}
|
||||
rp := int64(pb.RetentionPeriod.Round(time.Second) / time.Second)
|
||||
if rp > 0 {
|
||||
rules = append(rules, retentionRule{
|
||||
Type: "expire",
|
||||
EverySeconds: rp,
|
||||
})
|
||||
}
|
||||
|
||||
return &bucket{
|
||||
ID: pb.ID,
|
||||
OrgID: pb.OrgID,
|
||||
Type: pb.Type.String(),
|
||||
Name: pb.Name,
|
||||
Description: pb.Description,
|
||||
RetentionPolicyName: pb.RetentionPolicyName,
|
||||
RetentionRules: rules,
|
||||
CRUDLog: pb.CRUDLog,
|
||||
}
|
||||
}
|
||||
|
||||
// bucketUpdate is used for serialization/deserialization with retention rules.
|
||||
type bucketUpdate struct {
|
||||
Name *string `json:"name,omitempty"`
|
||||
Description *string `json:"description,omitempty"`
|
||||
RetentionRules []retentionRule `json:"retentionRules,omitempty"`
|
||||
}
|
||||
|
||||
func (b *bucketUpdate) OK() error {
|
||||
if len(b.RetentionRules) > 0 {
|
||||
_, err := b.RetentionRules[0].RetentionPeriod()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *bucketUpdate) toInfluxDB() *influxdb.BucketUpdate {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// For now, only use a single retention rule.
|
||||
var d time.Duration
|
||||
if len(b.RetentionRules) > 0 {
|
||||
d, _ = b.RetentionRules[0].RetentionPeriod()
|
||||
}
|
||||
|
||||
return &influxdb.BucketUpdate{
|
||||
Name: b.Name,
|
||||
Description: b.Description,
|
||||
RetentionPeriod: &d,
|
||||
}
|
||||
}
|
||||
|
||||
func newBucketUpdate(pb *influxdb.BucketUpdate) *bucketUpdate {
|
||||
if pb == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
up := &bucketUpdate{
|
||||
Name: pb.Name,
|
||||
Description: pb.Description,
|
||||
RetentionRules: []retentionRule{},
|
||||
}
|
||||
|
||||
if pb.RetentionPeriod != nil {
|
||||
d := int64((*pb.RetentionPeriod).Round(time.Second) / time.Second)
|
||||
up.RetentionRules = append(up.RetentionRules, retentionRule{
|
||||
Type: "expire",
|
||||
EverySeconds: d,
|
||||
})
|
||||
}
|
||||
return up
|
||||
}
|
||||
|
||||
type bucketResponse struct {
|
||||
bucket
|
||||
Links map[string]string `json:"links"`
|
||||
}
|
||||
|
||||
func NewBucketResponse(b *influxdb.Bucket) *bucketResponse {
|
||||
res := &bucketResponse{
|
||||
Links: map[string]string{
|
||||
"self": fmt.Sprintf("/api/v2/buckets/%s", b.ID),
|
||||
"org": fmt.Sprintf("/api/v2/orgs/%s", b.OrgID),
|
||||
"members": fmt.Sprintf("/api/v2/buckets/%s/members", b.ID),
|
||||
"owners": fmt.Sprintf("/api/v2/buckets/%s/owners", b.ID),
|
||||
"labels": fmt.Sprintf("/api/v2/buckets/%s/labels", b.ID),
|
||||
"write": fmt.Sprintf("/api/v2/write?org=%s&bucket=%s", b.OrgID, b.ID),
|
||||
},
|
||||
bucket: *newBucket(b),
|
||||
}
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
type bucketsResponse struct {
|
||||
Links *influxdb.PagingLinks `json:"links"`
|
||||
Buckets []*bucketResponse `json:"buckets"`
|
||||
}
|
||||
|
||||
func newBucketsResponse(ctx context.Context, opts influxdb.FindOptions, f influxdb.BucketFilter, bs []*influxdb.Bucket) *bucketsResponse {
|
||||
rs := make([]*bucketResponse, 0, len(bs))
|
||||
for _, b := range bs {
|
||||
rs = append(rs, NewBucketResponse(b))
|
||||
}
|
||||
return &bucketsResponse{
|
||||
Links: influxdb.NewPagingLinks(prefixBuckets, opts, f, len(bs)),
|
||||
Buckets: rs,
|
||||
}
|
||||
}
|
||||
|
||||
// handlePostBucket is the HTTP handler for the POST /api/v2/buckets route.
|
||||
func (h *BucketHandler) handlePostBucket(w http.ResponseWriter, r *http.Request) {
|
||||
var b postBucketRequest
|
||||
if err := h.api.DecodeJSON(r.Body, &b); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
if err := b.OK(); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
bucket := b.toInfluxDB()
|
||||
|
||||
if err := validBucketName(bucket); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.bucketSvc.CreateBucket(r.Context(), bucket); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Bucket created", zap.String("bucket", fmt.Sprint(bucket)))
|
||||
|
||||
h.api.Respond(w, http.StatusCreated, NewBucketResponse(bucket))
|
||||
}
|
||||
|
||||
type postBucketRequest struct {
|
||||
OrgID influxdb.ID `json:"orgID,omitempty"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
RetentionPolicyName string `json:"rp,omitempty"` // This to support v1 sources
|
||||
RetentionRules []retentionRule `json:"retentionRules"`
|
||||
}
|
||||
|
||||
func (b *postBucketRequest) OK() error {
|
||||
if !b.OrgID.Valid() {
|
||||
return &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "organization id must be provided",
|
||||
}
|
||||
}
|
||||
|
||||
// Only support a single retention period for the moment
|
||||
if len(b.RetentionRules) > 0 {
|
||||
if _, err := b.RetentionRules[0].RetentionPeriod(); err != nil {
|
||||
return &influxdb.Error{
|
||||
Code: influxdb.EUnprocessableEntity,
|
||||
Msg: err.Error(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// names starting with an underscore are reserved for system buckets
|
||||
if err := validBucketName(b.toInfluxDB()); err != nil {
|
||||
return &influxdb.Error{
|
||||
Code: influxdb.EUnprocessableEntity,
|
||||
Msg: err.Error(),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b postBucketRequest) toInfluxDB() *influxdb.Bucket {
|
||||
// Only support a single retention period for the moment
|
||||
var dur time.Duration
|
||||
if len(b.RetentionRules) > 0 {
|
||||
dur, _ = b.RetentionRules[0].RetentionPeriod()
|
||||
}
|
||||
|
||||
return &influxdb.Bucket{
|
||||
OrgID: b.OrgID,
|
||||
Description: b.Description,
|
||||
Name: b.Name,
|
||||
Type: influxdb.BucketTypeUser,
|
||||
RetentionPolicyName: b.RetentionPolicyName,
|
||||
RetentionPeriod: dur,
|
||||
}
|
||||
}
|
||||
|
||||
// handleGetBucket is the HTTP handler for the GET /api/v2/buckets/:id route.
|
||||
func (h *BucketHandler) handleGetBucket(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
b, err := h.bucketSvc.FindBucketByID(ctx, *id)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Debug("Bucket retrieved", zap.String("bucket", fmt.Sprint(b)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, NewBucketResponse(b))
|
||||
}
|
||||
|
||||
// handleDeleteBucket is the HTTP handler for the DELETE /api/v2/buckets/:id route.
|
||||
func (h *BucketHandler) handleDeleteBucket(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.bucketSvc.DeleteBucket(r.Context(), *id); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Debug("Bucket deleted", zap.String("bucketID", id.String()))
|
||||
|
||||
h.api.Respond(w, http.StatusNoContent, nil)
|
||||
}
|
||||
|
||||
// handleGetBuckets is the HTTP handler for the GET /api/v2/buckets route.
|
||||
func (h *BucketHandler) handleGetBuckets(w http.ResponseWriter, r *http.Request) {
|
||||
bucketsRequest, err := decodeGetBucketsRequest(r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
bs, _, err := h.bucketSvc.FindBuckets(r.Context(), bucketsRequest.filter, bucketsRequest.opts)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Buckets retrieved", zap.String("buckets", fmt.Sprint(bs)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newBucketsResponse(r.Context(), bucketsRequest.opts, bucketsRequest.filter, bs))
|
||||
}
|
||||
|
||||
type getBucketsRequest struct {
|
||||
filter influxdb.BucketFilter
|
||||
opts influxdb.FindOptions
|
||||
}
|
||||
|
||||
func decodeGetBucketsRequest(r *http.Request) (*getBucketsRequest, error) {
|
||||
qp := r.URL.Query()
|
||||
req := &getBucketsRequest{}
|
||||
|
||||
opts, err := decodeFindOptions(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req.opts = *opts
|
||||
|
||||
if orgID := qp.Get("orgID"); orgID != "" {
|
||||
id, err := influxdb.IDFromString(orgID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.filter.OrganizationID = id
|
||||
}
|
||||
|
||||
if org := qp.Get("org"); org != "" {
|
||||
req.filter.Org = &org
|
||||
}
|
||||
|
||||
if name := qp.Get("name"); name != "" {
|
||||
req.filter.Name = &name
|
||||
}
|
||||
|
||||
if bucketID := qp.Get("id"); bucketID != "" {
|
||||
id, err := influxdb.IDFromString(bucketID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.filter.ID = id
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// handlePatchBucket is the HTTP handler for the PATCH /api/v2/buckets route.
|
||||
func (h *BucketHandler) handlePatchBucket(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
var reqBody bucketUpdate
|
||||
if err := h.api.DecodeJSON(r.Body, &reqBody); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if reqBody.Name != nil {
|
||||
b, err := h.bucketSvc.FindBucketByID(r.Context(), *id)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
b.Name = *reqBody.Name
|
||||
if err := validBucketName(b); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
b, err := h.bucketSvc.UpdateBucket(r.Context(), *id, *reqBody.toInfluxDB())
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Debug("Bucket updated", zap.String("bucket", fmt.Sprint(b)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, NewBucketResponse(b))
|
||||
}
|
||||
|
||||
// validBucketName reports any errors with bucket names
|
||||
func validBucketName(bucket *influxdb.Bucket) error {
|
||||
// names starting with an underscore are reserved for system buckets
|
||||
if strings.HasPrefix(bucket.Name, "_") && bucket.Type != influxdb.BucketTypeSystem {
|
||||
return &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Op: "http/bucket",
|
||||
Msg: fmt.Sprintf("bucket name %s is invalid. Buckets may not start with underscore", bucket.Name),
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
package tenant_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/influxdata/influxdb"
|
||||
ihttp "github.com/influxdata/influxdb/http"
|
||||
"github.com/influxdata/influxdb/tenant"
|
||||
itesting "github.com/influxdata/influxdb/testing"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
func initBucketHttpService(f itesting.BucketFields, t *testing.T) (influxdb.BucketService, string, func()) {
|
||||
t.Helper()
|
||||
|
||||
s, stCloser, err := NewTestInmemStore(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
storage, err := tenant.NewStore(s)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc := tenant.NewService(storage)
|
||||
|
||||
ctx := context.Background()
|
||||
for _, o := range f.Organizations {
|
||||
if err := svc.CreateOrganization(ctx, o); err != nil {
|
||||
t.Fatalf("failed to populate organizations")
|
||||
}
|
||||
}
|
||||
for _, b := range f.Buckets {
|
||||
if err := svc.CreateBucket(ctx, b); err != nil {
|
||||
t.Fatalf("failed to populate buckets")
|
||||
}
|
||||
}
|
||||
|
||||
handler := tenant.NewHTTPBucketHandler(zaptest.NewLogger(t), svc, nil, nil)
|
||||
r := chi.NewRouter()
|
||||
r.Mount(handler.Prefix(), handler)
|
||||
server := httptest.NewServer(r)
|
||||
httpClient, err := ihttp.NewHTTPClient(server.URL, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
client := tenant.BucketClientService{
|
||||
Client: httpClient,
|
||||
}
|
||||
|
||||
return &client, "http_tenant", func() {
|
||||
server.Close()
|
||||
stCloser()
|
||||
}
|
||||
}
|
||||
|
||||
func TestBucketService(t *testing.T) {
|
||||
itesting.BucketService(initBucketHttpService, t, itesting.WithoutHooks())
|
||||
}
|
||||
|
|
@ -0,0 +1,191 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/go-chi/chi/middleware"
|
||||
"github.com/influxdata/influxdb"
|
||||
kithttp "github.com/influxdata/influxdb/kit/transport/http"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// OrgHandler represents an HTTP API handler for organizations.
|
||||
type OrgHandler struct {
|
||||
chi.Router
|
||||
api *kithttp.API
|
||||
log *zap.Logger
|
||||
orgSvc influxdb.OrganizationService
|
||||
}
|
||||
|
||||
const (
|
||||
prefixOrganizations = "/api/v2/orgs"
|
||||
)
|
||||
|
||||
func (h *OrgHandler) Prefix() string {
|
||||
return prefixOrganizations
|
||||
}
|
||||
|
||||
// NewHTTPOrgHandler constructs a new http server.
|
||||
func NewHTTPOrgHandler(log *zap.Logger, orgService influxdb.OrganizationService, urm http.Handler, labelHandler http.Handler, secretHandler http.Handler) *OrgHandler {
|
||||
svr := &OrgHandler{
|
||||
api: kithttp.NewAPI(kithttp.WithLog(log)),
|
||||
log: log,
|
||||
orgSvc: orgService,
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Use(
|
||||
middleware.Recoverer,
|
||||
middleware.RequestID,
|
||||
middleware.RealIP,
|
||||
)
|
||||
|
||||
r.Route("/", func(r chi.Router) {
|
||||
r.Post("/", svr.handlePostOrg)
|
||||
r.Get("/", svr.handleGetOrgs)
|
||||
|
||||
r.Route("/{id}", func(r chi.Router) {
|
||||
r.Get("/", svr.handleGetOrg)
|
||||
r.Patch("/", svr.handlePatchOrg)
|
||||
r.Delete("/", svr.handleDeleteOrg)
|
||||
|
||||
// mount embedded resources
|
||||
r.Mount("/members", urm)
|
||||
r.Mount("/owners", urm)
|
||||
r.Mount("/labels", labelHandler)
|
||||
r.Mount("/secrets", secretHandler)
|
||||
})
|
||||
})
|
||||
|
||||
svr.Router = r
|
||||
return svr
|
||||
}
|
||||
|
||||
type orgResponse struct {
|
||||
influxdb.Organization
|
||||
}
|
||||
|
||||
func newOrgResponse(o influxdb.Organization) orgResponse {
|
||||
return orgResponse{
|
||||
Organization: o,
|
||||
}
|
||||
}
|
||||
|
||||
type orgsResponse struct {
|
||||
Organizations []orgResponse `json:"orgs"`
|
||||
}
|
||||
|
||||
func newOrgsResponse(orgs []*influxdb.Organization) *orgsResponse {
|
||||
res := orgsResponse{
|
||||
Organizations: []orgResponse{},
|
||||
}
|
||||
for _, org := range orgs {
|
||||
res.Organizations = append(res.Organizations, newOrgResponse(*org))
|
||||
}
|
||||
return &res
|
||||
}
|
||||
|
||||
// handlePostOrg is the HTTP handler for the POST /api/v2/orgs route.
|
||||
func (h *OrgHandler) handlePostOrg(w http.ResponseWriter, r *http.Request) {
|
||||
var org influxdb.Organization
|
||||
if err := h.api.DecodeJSON(r.Body, &org); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.orgSvc.CreateOrganization(r.Context(), &org); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Debug("Org created", zap.String("org", fmt.Sprint(org)))
|
||||
|
||||
h.api.Respond(w, http.StatusCreated, newOrgResponse(org))
|
||||
}
|
||||
|
||||
// handleGetOrg is the HTTP handler for the GET /api/v2/orgs/:id route.
|
||||
func (h *OrgHandler) handleGetOrg(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
org, err := h.orgSvc.FindOrganizationByID(r.Context(), *id)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Org retrieved", zap.String("org", fmt.Sprint(org)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newOrgResponse(*org))
|
||||
}
|
||||
|
||||
// handleGetOrgs is the HTTP handler for the GET /api/v2/orgs route.
|
||||
func (h *OrgHandler) handleGetOrgs(w http.ResponseWriter, r *http.Request) {
|
||||
var filter influxdb.OrganizationFilter
|
||||
qp := r.URL.Query()
|
||||
if name := qp.Get("org"); name != "" {
|
||||
filter.Name = &name
|
||||
}
|
||||
|
||||
if id := qp.Get("orgID"); id != "" {
|
||||
i, err := influxdb.IDFromString(id)
|
||||
if err == nil {
|
||||
filter.ID = i
|
||||
}
|
||||
}
|
||||
|
||||
orgs, _, err := h.orgSvc.FindOrganizations(r.Context(), filter)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Orgs retrieved", zap.String("org", fmt.Sprint(orgs)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newOrgsResponse(orgs))
|
||||
}
|
||||
|
||||
// handlePatchOrg is the HTTP handler for the PATH /api/v2/orgs route.
|
||||
func (h *OrgHandler) handlePatchOrg(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
var upd influxdb.OrganizationUpdate
|
||||
if err := h.api.DecodeJSON(r.Body, &upd); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
org, err := h.orgSvc.UpdateOrganization(r.Context(), *id, upd)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Org updated", zap.String("org", fmt.Sprint(org)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newOrgResponse(*org))
|
||||
}
|
||||
|
||||
// handleDeleteOrganization is the HTTP handler for the DELETE /api/v2/orgs/:id route.
|
||||
func (h *OrgHandler) handleDeleteOrg(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := influxdb.IDFromString(chi.URLParam(r, "id"))
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
if err := h.orgSvc.DeleteOrganization(ctx, *id); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Org deleted", zap.String("orgID", fmt.Sprint(id)))
|
||||
|
||||
h.api.Respond(w, http.StatusNoContent, nil)
|
||||
}
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
package tenant_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/influxdata/influxdb"
|
||||
"github.com/influxdata/influxdb/http"
|
||||
"github.com/influxdata/influxdb/tenant"
|
||||
itesting "github.com/influxdata/influxdb/testing"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
func initHttpOrgService(f itesting.OrganizationFields, t *testing.T) (influxdb.OrganizationService, string, func()) {
|
||||
t.Helper()
|
||||
|
||||
s, stCloser, err := NewTestInmemStore(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
storage, err := tenant.NewStore(s)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc := tenant.NewService(storage)
|
||||
ctx := context.Background()
|
||||
for _, o := range f.Organizations {
|
||||
if err := svc.CreateOrganization(ctx, o); err != nil {
|
||||
t.Fatalf("failed to populate organizations")
|
||||
}
|
||||
}
|
||||
|
||||
handler := tenant.NewHTTPOrgHandler(zaptest.NewLogger(t), svc, nil, nil, nil)
|
||||
r := chi.NewRouter()
|
||||
r.Mount(handler.Prefix(), handler)
|
||||
server := httptest.NewServer(r)
|
||||
httpClient, err := http.NewHTTPClient(server.URL, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
orgClient := tenant.OrgClientService{
|
||||
Client: httpClient,
|
||||
}
|
||||
|
||||
return &orgClient, "http_tenant", func() {
|
||||
server.Close()
|
||||
stCloser()
|
||||
}
|
||||
}
|
||||
|
||||
func TestOrgService(t *testing.T) {
|
||||
itesting.OrganizationService(initHttpOrgService, t)
|
||||
}
|
||||
|
|
@ -0,0 +1,464 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/go-chi/chi/middleware"
|
||||
"github.com/influxdata/influxdb"
|
||||
icontext "github.com/influxdata/influxdb/context"
|
||||
kithttp "github.com/influxdata/influxdb/kit/transport/http"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// UserHandler represents an HTTP API handler for users.
|
||||
type UserHandler struct {
|
||||
chi.Router
|
||||
api *kithttp.API
|
||||
log *zap.Logger
|
||||
userSvc influxdb.UserService
|
||||
passwordSvc influxdb.PasswordsService
|
||||
}
|
||||
|
||||
const (
|
||||
prefixUsers = "/api/v2/users"
|
||||
prefixMe = "/api/v2/me"
|
||||
)
|
||||
|
||||
// NewHTTPUserHandler constructs a new http server.
|
||||
func NewHTTPUserHandler(log *zap.Logger, userService influxdb.UserService, passwordService influxdb.PasswordsService) *UserHandler {
|
||||
svr := &UserHandler{
|
||||
api: kithttp.NewAPI(kithttp.WithLog(log)),
|
||||
log: log,
|
||||
userSvc: userService,
|
||||
passwordSvc: passwordService,
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Use(
|
||||
middleware.Recoverer,
|
||||
middleware.RequestID,
|
||||
middleware.RealIP,
|
||||
)
|
||||
|
||||
// RESTy routes for "articles" resource
|
||||
r.Route("/", func(r chi.Router) {
|
||||
r.Post("/", svr.handlePostUser)
|
||||
r.Get("/", svr.handleGetUsers)
|
||||
r.Put("/password", svr.handlePutUserPassword)
|
||||
|
||||
r.Route("/{id}", func(r chi.Router) {
|
||||
r.Get("/", svr.handleGetUser)
|
||||
r.Patch("/", svr.handlePatchUser)
|
||||
r.Delete("/", svr.handleDeleteUser)
|
||||
r.Put("/password", svr.handlePutUserPassword)
|
||||
r.Post("/password", svr.handlePostUserPassword)
|
||||
})
|
||||
})
|
||||
|
||||
svr.Router = r
|
||||
return svr
|
||||
}
|
||||
|
||||
type resourceHandler struct {
|
||||
prefix string
|
||||
*UserHandler
|
||||
}
|
||||
|
||||
func (h *UserHandler) MeResourceHandler() *resourceHandler {
|
||||
return &resourceHandler{prefix: prefixMe, UserHandler: h}
|
||||
}
|
||||
|
||||
func (h *UserHandler) UserResourceHandler() *resourceHandler {
|
||||
return &resourceHandler{prefix: prefixUsers, UserHandler: h}
|
||||
}
|
||||
|
||||
type passwordSetRequest struct {
|
||||
Password string `json:"password"`
|
||||
}
|
||||
|
||||
// handlePutPassword is the HTTP handler for the PUT /api/v2/users/:id/password
|
||||
func (h *UserHandler) handlePostUserPassword(w http.ResponseWriter, r *http.Request) {
|
||||
var body passwordSetRequest
|
||||
err := json.NewDecoder(r.Body).Decode(&body)
|
||||
if err != nil {
|
||||
h.api.Err(w, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Err: err,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
param := chi.URLParam(r, "id")
|
||||
userID, err := influxdb.IDFromString(param)
|
||||
if err != nil {
|
||||
h.api.Err(w, &influxdb.Error{
|
||||
Msg: "invalid user ID provided in route",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
err = h.passwordSvc.SetPassword(r.Context(), *userID, body.Password)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (h *UserHandler) putPassword(ctx context.Context, w http.ResponseWriter, r *http.Request) (username string, err error) {
|
||||
req, err := decodePasswordResetRequest(r)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
param := chi.URLParam(r, "id")
|
||||
userID, err := influxdb.IDFromString(param)
|
||||
if err != nil {
|
||||
h.api.Err(w, &influxdb.Error{
|
||||
Msg: "invalid user ID provided in route",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
err = h.passwordSvc.CompareAndSetPassword(ctx, *userID, req.PasswordOld, req.PasswordNew)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return req.Username, nil
|
||||
}
|
||||
|
||||
// handlePutPassword is the HTTP handler for the PUT /api/v2/users/:id/password
|
||||
func (h *UserHandler) handlePutUserPassword(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
_, err := h.putPassword(ctx, w, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("User password updated")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
type passwordResetRequest struct {
|
||||
Username string
|
||||
PasswordOld string
|
||||
PasswordNew string
|
||||
}
|
||||
|
||||
type passwordResetRequestBody struct {
|
||||
Password string `json:"password"`
|
||||
}
|
||||
|
||||
func decodePasswordResetRequest(r *http.Request) (*passwordResetRequest, error) {
|
||||
u, o, ok := r.BasicAuth()
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("invalid basic auth")
|
||||
}
|
||||
|
||||
pr := new(passwordResetRequestBody)
|
||||
err := json.NewDecoder(r.Body).Decode(pr)
|
||||
if err != nil {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
return &passwordResetRequest{
|
||||
Username: u,
|
||||
PasswordOld: o,
|
||||
PasswordNew: pr.Password,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// handlePostUser is the HTTP handler for the POST /api/v2/users route.
|
||||
func (h *UserHandler) handlePostUser(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
req, err := decodePostUserRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if req.User.Status == "" {
|
||||
req.User.Status = influxdb.Active
|
||||
}
|
||||
|
||||
if err := h.userSvc.CreateUser(ctx, req.User); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("User created", zap.String("user", fmt.Sprint(req.User)))
|
||||
|
||||
h.api.Respond(w, http.StatusCreated, newUserResponse(req.User))
|
||||
}
|
||||
|
||||
type postUserRequest struct {
|
||||
User *influxdb.User
|
||||
}
|
||||
|
||||
func decodePostUserRequest(ctx context.Context, r *http.Request) (*postUserRequest, error) {
|
||||
b := &influxdb.User{}
|
||||
if err := json.NewDecoder(r.Body).Decode(b); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &postUserRequest{
|
||||
User: b,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// handleGetMe is the HTTP handler for the GET /api/v2/me.
|
||||
func (h *UserHandler) handleGetMe(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
|
||||
a, err := icontext.GetAuthorizer(ctx)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
id := a.GetUserID()
|
||||
user, err := h.userSvc.FindUserByID(ctx, id)
|
||||
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newUserResponse(user))
|
||||
}
|
||||
|
||||
// handleGetUser is the HTTP handler for the GET /api/v2/users/:id route.
|
||||
func (h *UserHandler) handleGetUser(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
req, err := decodeGetUserRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
b, err := h.userSvc.FindUserByID(ctx, req.UserID)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("User retrieved", zap.String("user", fmt.Sprint(b)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newUserResponse(b))
|
||||
}
|
||||
|
||||
type getUserRequest struct {
|
||||
UserID influxdb.ID
|
||||
}
|
||||
|
||||
func decodeGetUserRequest(ctx context.Context, r *http.Request) (*getUserRequest, error) {
|
||||
id := chi.URLParam(r, "id")
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var i influxdb.ID
|
||||
if err := i.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req := &getUserRequest{
|
||||
UserID: i,
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// handleDeleteUser is the HTTP handler for the DELETE /api/v2/users/:id route.
|
||||
func (h *UserHandler) handleDeleteUser(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
req, err := decodeDeleteUserRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.userSvc.DeleteUser(ctx, req.UserID); err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("User deleted", zap.String("userID", fmt.Sprint(req.UserID)))
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
type deleteUserRequest struct {
|
||||
UserID influxdb.ID
|
||||
}
|
||||
|
||||
func decodeDeleteUserRequest(ctx context.Context, r *http.Request) (*deleteUserRequest, error) {
|
||||
id := chi.URLParam(r, "id")
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var i influxdb.ID
|
||||
if err := i.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &deleteUserRequest{
|
||||
UserID: i,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type usersResponse struct {
|
||||
Links map[string]string `json:"links"`
|
||||
Users []*userResponse `json:"users"`
|
||||
}
|
||||
|
||||
func (us usersResponse) ToInfluxdb() []*influxdb.User {
|
||||
users := make([]*influxdb.User, len(us.Users))
|
||||
for i := range us.Users {
|
||||
users[i] = &us.Users[i].User
|
||||
}
|
||||
return users
|
||||
}
|
||||
|
||||
func newUsersResponse(users []*influxdb.User) *usersResponse {
|
||||
res := usersResponse{
|
||||
Links: map[string]string{
|
||||
"self": "/api/v2/users",
|
||||
},
|
||||
Users: []*userResponse{},
|
||||
}
|
||||
for _, user := range users {
|
||||
res.Users = append(res.Users, newUserResponse(user))
|
||||
}
|
||||
return &res
|
||||
}
|
||||
|
||||
// userResponse is the response of user
|
||||
type userResponse struct {
|
||||
Links map[string]string `json:"links"`
|
||||
influxdb.User
|
||||
}
|
||||
|
||||
func newUserResponse(u *influxdb.User) *userResponse {
|
||||
return &userResponse{
|
||||
Links: map[string]string{
|
||||
"self": fmt.Sprintf("/api/v2/users/%s", u.ID),
|
||||
},
|
||||
User: *u,
|
||||
}
|
||||
}
|
||||
|
||||
// handleGetUsers is the HTTP handler for the GET /api/v2/users route.
|
||||
func (h *UserHandler) handleGetUsers(w http.ResponseWriter, r *http.Request) {
|
||||
// because this is a mounted path in both the /users and the /me route
|
||||
// we can get a me request through this handler
|
||||
if strings.Contains(r.URL.Path, prefixMe) {
|
||||
h.handleGetMe(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
req, err := decodeGetUsersRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
users, _, err := h.userSvc.FindUsers(ctx, req.filter)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Users retrieved", zap.String("users", fmt.Sprint(users)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newUsersResponse(users))
|
||||
}
|
||||
|
||||
type getUsersRequest struct {
|
||||
filter influxdb.UserFilter
|
||||
}
|
||||
|
||||
func decodeGetUsersRequest(ctx context.Context, r *http.Request) (*getUsersRequest, error) {
|
||||
qp := r.URL.Query()
|
||||
req := &getUsersRequest{}
|
||||
|
||||
if userID := qp.Get("id"); userID != "" {
|
||||
id, err := influxdb.IDFromString(userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.filter.ID = id
|
||||
}
|
||||
|
||||
if name := qp.Get("name"); name != "" {
|
||||
req.filter.Name = &name
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// handlePatchUser is the HTTP handler for the PATCH /api/v2/users/:id route.
|
||||
func (h *UserHandler) handlePatchUser(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
req, err := decodePatchUserRequest(ctx, r)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
b, err := h.userSvc.UpdateUser(ctx, req.UserID, req.Update)
|
||||
if err != nil {
|
||||
h.api.Err(w, err)
|
||||
return
|
||||
}
|
||||
h.log.Debug("Users updated", zap.String("user", fmt.Sprint(b)))
|
||||
|
||||
h.api.Respond(w, http.StatusOK, newUserResponse(b))
|
||||
}
|
||||
|
||||
type patchUserRequest struct {
|
||||
Update influxdb.UserUpdate
|
||||
UserID influxdb.ID
|
||||
}
|
||||
|
||||
func decodePatchUserRequest(ctx context.Context, r *http.Request) (*patchUserRequest, error) {
|
||||
id := chi.URLParam(r, "id")
|
||||
if id == "" {
|
||||
return nil, &influxdb.Error{
|
||||
Code: influxdb.EInvalid,
|
||||
Msg: "url missing id",
|
||||
}
|
||||
}
|
||||
|
||||
var i influxdb.ID
|
||||
if err := i.DecodeFromString(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var upd influxdb.UserUpdate
|
||||
if err := json.NewDecoder(r.Body).Decode(&upd); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := upd.Valid(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &patchUserRequest{
|
||||
Update: upd,
|
||||
UserID: i,
|
||||
}, nil
|
||||
}
|
||||
|
|
@ -0,0 +1,61 @@
|
|||
package tenant_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
platform "github.com/influxdata/influxdb"
|
||||
ihttp "github.com/influxdata/influxdb/http"
|
||||
"github.com/influxdata/influxdb/tenant"
|
||||
platformtesting "github.com/influxdata/influxdb/testing"
|
||||
"go.uber.org/zap/zaptest"
|
||||
)
|
||||
|
||||
func initHttpUserService(f platformtesting.UserFields, t *testing.T) (platform.UserService, string, func()) {
|
||||
t.Helper()
|
||||
|
||||
s, stCloser, err := NewTestInmemStore(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
storage, err := tenant.NewStore(s)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc := tenant.NewService(storage)
|
||||
|
||||
ctx := context.Background()
|
||||
for _, u := range f.Users {
|
||||
if err := svc.CreateUser(ctx, u); err != nil {
|
||||
t.Fatalf("failed to populate users")
|
||||
}
|
||||
}
|
||||
|
||||
handler := tenant.NewHTTPUserHandler(zaptest.NewLogger(t), svc, svc)
|
||||
r := chi.NewRouter()
|
||||
r.Mount("/api/v2/users", handler)
|
||||
r.Mount("/api/v2/me", handler)
|
||||
server := httptest.NewServer(r)
|
||||
|
||||
httpClient, err := ihttp.NewHTTPClient(server.URL, "", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
client := tenant.UserClientService{
|
||||
Client: httpClient,
|
||||
}
|
||||
|
||||
return &client, "http_tenant", func() {
|
||||
server.Close()
|
||||
stCloser()
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserService(t *testing.T) {
|
||||
t.Parallel()
|
||||
platformtesting.UserService(initHttpUserService, t)
|
||||
}
|
||||
Loading…
Reference in New Issue