Skip to content

[wip]add cache for proxy api #263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

[wip]add cache for proxy api #263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b306ad3
add cache for proxy api
tiancandevloper Mar 27, 2023
ef47ce2
update code
tiancandevloper Mar 28, 2023
d4454d5
update fmt
tiancandevloper Mar 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 116 additions & 13 deletions pkg/apiserver/cubeapi/resourcemanage/handle/proxy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,14 @@ import (

"github.com/gin-gonic/gin"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/kubecube-io/kubecube/pkg/apiserver/cubeapi/resourcemanage/resources"
"github.com/kubecube-io/kubecube/pkg/belongs"
"github.com/kubecube-io/kubecube/pkg/clog"
"github.com/kubecube-io/kubecube/pkg/conversion"
Expand All @@ -40,6 +45,7 @@ import (
"github.com/kubecube-io/kubecube/pkg/utils/errcode"
"github.com/kubecube-io/kubecube/pkg/utils/filter"
"github.com/kubecube-io/kubecube/pkg/utils/page"
"github.com/kubecube-io/kubecube/pkg/utils/path"
requestutil "github.com/kubecube-io/kubecube/pkg/utils/request"
"github.com/kubecube-io/kubecube/pkg/utils/response"
"github.com/kubecube-io/kubecube/pkg/utils/selector"
Expand Down Expand Up @@ -146,42 +152,139 @@ func (h *ProxyHandler) ProxyHandle(c *gin.Context) {
if len(username) == 0 {
clog.Warn("username is empty")
}
condition := parseQueryParams(c)
converterContext := filter.ConverterContext{}
c.Request.Header.Set(constants.ImpersonateUserKey, username)
internalCluster, err := multicluster.Interface().Get(cluster)
if err != nil {
clog.Error(err.Error())
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
transport, err := multicluster.Interface().GetTransport(cluster)
allowed, err := belongs.RelationshipDetermine(context.Background(), internalCluster.Client, proxyUrl, username)
if err != nil {
clog.Warn(err.Error())
} else if !allowed {
response.FailReturn(c, errcode.ForbiddenErr)
return
}

if c.Request.Method == http.MethodGet {
h.cacheProxy(c, proxyUrl, username, internalCluster)
return
}
h.directProxy(c, proxyUrl, username, internalCluster)
}

func (h *ProxyHandler) cacheProxy(c *gin.Context, proxyUrl string, username string, internalCluster *multicluster.InternalCluster) {
parse, err := path.Parse(proxyUrl)
if err != nil {
clog.Error(err.Error())
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
_, _, gvr, err := conversion.ParseURL(proxyUrl)
if len(parse.SubResource) != 0 {
h.directProxy(c, proxyUrl, username, internalCluster)
return
}
field := c.Query("fieldSelector")
if len(field) != 0 {
h.directProxy(c, proxyUrl, username, internalCluster)
return
}
access := resources.NewSimpleAccess(internalCluster.Name, username, parse.Namespace)
if allow := access.AccessAllow(parse.Gvr.Group, parse.Gvr.Resource, "get"); !allow {
response.FailReturn(c, errcode.ForbiddenErr)
return
}
gvk, err := conversion.Gvr2Gvk(internalCluster.Client.RESTMapper(), &parse.Gvr)
if err != nil {
clog.Error(err.Error())
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
getFunc := func(ns string) (*unstructured.Unstructured, error) {
u := &unstructured.Unstructured{}
u.SetGroupVersionKind(*gvk)
err = internalCluster.Client.Cache().Get(context.Background(), types.NamespacedName{
Name: parse.Name,
Namespace: ns,
}, u)
return u, err
}

listFunc := func(ns string, options ...client.ListOption) (*unstructured.Unstructured, error) {
u := &unstructured.UnstructuredList{}
if !strings.HasSuffix(gvk.Kind, "List") {
gvk.Kind = gvk.Kind + "List"
}
u.SetGroupVersionKind(*gvk)
if len(ns) != 0 {
options = append(options, client.InNamespace(ns))
}
label := c.Query("labelSelector")
if len(label) != 0 {
labelSelector, err := labels.Parse(label)
if err != nil {
return nil, err
}
options = append(options, client.MatchingLabelsSelector{Selector: labelSelector})
}
err = internalCluster.Client.Cache().List(context.Background(), u, options...)
if err != nil {
return nil, err
}
content := u.UnstructuredContent()
res := &unstructured.Unstructured{Object: content}
return res, nil
}

needConvert, convertedObj, convertedUrl, err := h.tryVersionConvert(cluster, proxyUrl, c.Request)
if parse.IsNamespaced {
if len(parse.Namespace) == 0 {
parse.Namespace = "default"
}
}

var u *unstructured.Unstructured
if len(parse.Name) == 0 {
u, err = listFunc(parse.Namespace)
if err != nil {
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
} else {
u, err = getFunc(parse.Namespace)
if err != nil {
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
}
condition := parseQueryParams(c)
total, err := filter.GetEmptyFilter().FilterObjectList(u, condition)
u.Object["total"] = total
response.SuccessReturn(c, u)
}

func (h *ProxyHandler) directProxy(c *gin.Context, proxyUrl string, username string, internalCluster *multicluster.InternalCluster) {
_, _, gvr, err := conversion.ParseURL(proxyUrl)
if err != nil {
clog.Error(err.Error())
response.FailReturn(c, errcode.InternalServerError)
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
converterContext := filter.ConverterContext{}
c.Request.Header.Set(constants.ImpersonateUserKey, username)

allowed, err := belongs.RelationshipDetermine(context.Background(), internalCluster.Client, proxyUrl, username)
transport, err := multicluster.Interface().GetTransport(internalCluster.Name)
if err != nil {
clog.Warn(err.Error())
} else if !allowed {
response.FailReturn(c, errcode.ForbiddenErr)
clog.Error(err.Error())
response.FailReturn(c, errcode.CustomReturn(http.StatusBadRequest, err.Error()))
return
}
needConvert, convertedObj, convertedUrl, err := h.tryVersionConvert(internalCluster.Name, proxyUrl, c.Request)
if err != nil {
clog.Error(err.Error())
response.FailReturn(c, errcode.InternalServerError)
return
}
condition := parseQueryParams(c)

// create director
director := func(req *http.Request) {
Expand Down Expand Up @@ -269,7 +372,7 @@ func (h *ProxyHandler) ProxyHandle(c *gin.Context) {

errorHandler := func(resp http.ResponseWriter, req *http.Request, err error) {
if err != nil {
clog.Warn("cluster %s url %s proxy fail, %v", cluster, proxyUrl, err)
clog.Warn("cluster %s url %s proxy fail, %v", internalCluster.Name, proxyUrl, err)
response.FailReturn(c, errcode.ServerErr)
return
}
Expand All @@ -284,7 +387,7 @@ func (h *ProxyHandler) ProxyHandle(c *gin.Context) {
return
}

converter, _ := h.converter.GetVersionConvert(cluster)
converter, _ := h.converter.GetVersionConvert(internalCluster.Name)
converterContext = filter.ConverterContext{
EnableConvert: true,
Converter: converter,
Expand Down
18 changes: 18 additions & 0 deletions pkg/utils/path/parse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ type ResourceIdentity struct {
IsCoreApi bool
IsNamespaced bool
Gvr schema.GroupVersionResource
SubResource string
}

// Parse parse k8s api url into gvr
Expand Down Expand Up @@ -57,6 +58,7 @@ func Parse(url string) (*ResourceIdentity, error) {
switch {
case isCoreApi && isNamespaced:
// like: /api/v1/namespaces/{namespace}/pods
// like: /api/v1/namespaces/{namespace}/pods/status
if len(ss) < 5 {
return nil, invalidUrlErr
}
Expand All @@ -66,8 +68,12 @@ func Parse(url string) (*ResourceIdentity, error) {
if len(ss) > 5 {
ri.Name = ss[5]
}
if len(ss) > 6 {
ri.SubResource = ss[6]
}
case isCoreApi && !isNamespaced:
// like: /api/v1/namespaces/{name}
// like: /api/v1/namespaces/{name}/status
if len(ss) < 3 {
return nil, invalidUrlErr
}
Expand All @@ -76,19 +82,28 @@ func Parse(url string) (*ResourceIdentity, error) {
if len(ss) > 3 {
ri.Name = ss[3]
}
if len(ss) > 4 {
ri.Name = ss[4]
}
case isNonCoreApi && isNamespaced:
// like: /apis/batch/v1/namespaces/{namespace}/jobs
// like: /apis/batch/v1/namespaces/{namespace}/jobs/status
if len(ss) < 6 {
return nil, invalidUrlErr
}
ri.Gvr.Group = ss[1]
ri.Gvr.Version = ss[2]
ri.Namespace = ss[4]
ri.Gvr.Resource = ss[5]
if len(ss) > 6 {
ri.Name = ss[6]
}
if len(ss) > 7 {
ri.SubResource = ss[7]
}
case isNonCoreApi && !isNamespaced:
// like: /apis/rbac.authorization.k8s.io/v1/clusterroles/{name}
// like: /apis/rbac.authorization.k8s.io/v1/clusterroles/{name}/status
if len(ss) < 4 {
return nil, invalidUrlErr
}
Expand All @@ -98,6 +113,9 @@ func Parse(url string) (*ResourceIdentity, error) {
if len(ss) > 4 {
ri.Name = ss[4]
}
if len(ss) > 5 {
ri.SubResource = ss[5]
}
default:
return nil, invalidUrlErr
}
Expand Down