Support metric label selector for custom metrics

pkg/apiserver/cmapis.go

@@ -32,7 +32,7 @@ import (
 )
 
 func (s *CustomMetricsAdapterServer) InstallCustomMetricsAPI() error {
-	groupInfo := genericapiserver.NewDefaultAPIGroupInfo(custom_metrics.GroupName, Scheme, metav1.ParameterCodec, Codecs)
+	groupInfo := genericapiserver.NewDefaultAPIGroupInfo(custom_metrics.GroupName, Scheme, runtime.NewParameterCodec(Scheme), Codecs)
 
 	mainGroupVer := groupInfo.PrioritizedVersions[0]
 	preferredVersionForDiscovery := metav1.GroupVersionForDiscovery{

pkg/apiserver/endpoints/handlers/get.go

@@ -0,0 +1,175 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+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 handlers
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/endpoints/handlers"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	utiltrace "k8s.io/utils/trace"
+
+	cm_rest "github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/apiserver/registry/rest"
+)
+
+func ListResourceWithOptions(r cm_rest.ListerWithOptions, scope handlers.RequestScope) http.HandlerFunc {
+	return func(w http.ResponseWriter, req *http.Request) {
+		// For performance tracking purposes.
+		trace := utiltrace.New("List " + req.URL.Path)
+
+		namespace, err := scope.Namer.Namespace(req)
+		if err != nil {
+			writeError(&scope, err, w, req)
+			return
+		}
+
+		// Watches for single objects are routed to this function.
+		// Treat a name parameter the same as a field selector entry.
+		hasName := true
+		_, name, err := scope.Namer.Name(req)
+		if err != nil {
+			hasName = false
+		}
+
+		ctx := req.Context()
+		ctx = request.WithNamespace(ctx, namespace)
+
+		opts := metainternalversion.ListOptions{}
+		if err := metainternalversion.ParameterCodec.DecodeParameters(req.URL.Query(), scope.MetaGroupVersion, &opts); err != nil {
+			err = errors.NewBadRequest(err.Error())
+			writeError(&scope, err, w, req)
+			return
+		}
+
+		// transform fields
+		// TODO: DecodeParametersInto should do this.
+		if opts.FieldSelector != nil {
+			fn := func(label, value string) (newLabel, newValue string, err error) {
+				return scope.Convertor.ConvertFieldLabel(scope.Kind, label, value)
+			}
+			if opts.FieldSelector, err = opts.FieldSelector.Transform(fn); err != nil {
+				// TODO: allow bad request to set field causes based on query parameters
+				err = errors.NewBadRequest(err.Error())
+				writeError(&scope, err, w, req)
+				return
+			}
+		}
+
+		if hasName {
+			// metadata.name is the canonical internal name.
+			// SelectionPredicate will notice that this is a request for
+			// a single object and optimize the storage query accordingly.
+			nameSelector := fields.OneTermEqualSelector("metadata.name", name)
+
+			// Note that fieldSelector setting explicitly the "metadata.name"
+			// will result in reaching this branch (as the value of that field
+			// is propagated to requestInfo as the name parameter.
+			// That said, the allowed field selectors in this branch are:
+			// nil, fields.Everything and field selector matching metadata.name
+			// for our name.
+			if opts.FieldSelector != nil && !opts.FieldSelector.Empty() {
+				selectedName, ok := opts.FieldSelector.RequiresExactMatch("metadata.name")
+				if !ok || name != selectedName {
+					writeError(&scope, errors.NewBadRequest("fieldSelector metadata.name doesn't match requested name"), w, req)
+					return
+				}
+			} else {
+				opts.FieldSelector = nameSelector
+			}
+		}
+
+		// Log only long List requests (ignore Watch).
+		defer trace.LogIfLong(500 * time.Millisecond)
+		trace.Step("About to List from storage")
+		extraOpts, hasSubpath, subpathKey := r.NewListOptions()
+		if err := getRequestOptions(req, scope, extraOpts, hasSubpath, subpathKey, false); err != nil {
+			err = errors.NewBadRequest(err.Error())
+			writeError(&scope, err, w, req)
+			return
+		}
+		result, err := r.List(ctx, &opts, extraOpts)
+		if err != nil {
+			writeError(&scope, err, w, req)
+			return
+		}
+		trace.Step("Listing from storage done")
+		numberOfItems, err := setListSelfLink(result, ctx, req, scope.Namer)
+		if err != nil {
+			writeError(&scope, err, w, req)
+			return
+		}
+		trace.Step("Self-linking done")
+		// Ensure empty lists return a non-nil items slice
+		if numberOfItems == 0 && meta.IsListType(result) {
+			if err := meta.SetList(result, []runtime.Object{}); err != nil {
+				writeError(&scope, err, w, req)
+				return
+			}
+		}
+
+		responsewriters.WriteObject(http.StatusOK, scope.Kind.GroupVersion(), scope.Serializer, result, w, req)
+		trace.Step(fmt.Sprintf("Writing http response done (%d items)", numberOfItems))
+	}
+}
+
+// getRequestOptions parses out options and can include path information.  The path information shouldn't include the subresource.
+func getRequestOptions(req *http.Request, scope handlers.RequestScope, into runtime.Object, hasSubpath bool, subpathKey string, isSubresource bool) error {
+	if into == nil {
+		return nil
+	}
+
+	query := req.URL.Query()
+	if hasSubpath {
+		newQuery := make(url.Values)
+		for k, v := range query {
+			newQuery[k] = v
+		}
+
+		ctx := req.Context()
+		requestInfo, _ := request.RequestInfoFrom(ctx)
+		startingIndex := 2
+		if isSubresource {
+			startingIndex = 3
+		}
+
+		p := strings.Join(requestInfo.Parts[startingIndex:], "/")
+
+		// ensure non-empty subpaths correctly reflect a leading slash
+		if len(p) > 0 && !strings.HasPrefix(p, "/") {
+			p = "/" + p
+		}
+
+		// ensure subpaths correctly reflect the presence of a trailing slash on the original request
+		if strings.HasSuffix(requestInfo.Path, "/") && !strings.HasSuffix(p, "/") {
+			p += "/"
+		}
+
+		newQuery[subpathKey] = []string{p}
+		query = newQuery
+	}
+	return scope.ParameterCodec.DecodeParameters(query, scope.Kind.GroupVersion(), into)
+}

pkg/apiserver/endpoints/handlers/rest.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+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 handlers
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/endpoints/handlers"
+	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/klog"
+)
+
+func writeError(scope *handlers.RequestScope, err error, w http.ResponseWriter, req *http.Request) {
+	responsewriters.ErrorNegotiated(err, scope.Serializer, scope.Kind.GroupVersion(), w, req)
+}
+
+// setSelfLink sets the self link of an object (or the child items in a list) to the base URL of the request
+// plus the path and query generated by the provided linkFunc
+func setSelfLink(obj runtime.Object, requestInfo *request.RequestInfo, namer handlers.ScopeNamer) error {
+	// TODO: SelfLink generation should return a full URL?
+	uri, err := namer.GenerateLink(requestInfo, obj)
+	if err != nil {
+		return nil
+	}
+
+	return namer.SetSelfLink(obj, uri)
+}
+
+// setListSelfLink sets the self link of a list to the base URL, then sets the self links
+// on all child objects returned. Returns the number of items in the list.
+func setListSelfLink(obj runtime.Object, ctx context.Context, req *http.Request, namer handlers.ScopeNamer) (int, error) {
+	if !meta.IsListType(obj) {
+		return 0, nil
+	}
+
+	uri, err := namer.GenerateListLink(req)
+	if err != nil {
+		return 0, err
+	}
+	if err := namer.SetSelfLink(obj, uri); err != nil {
+		klog.V(4).Infof("Unable to set self link on object: %v", err)
+	}
+	requestInfo, ok := request.RequestInfoFrom(ctx)
+	if !ok {
+		return 0, fmt.Errorf("missing requestInfo")
+	}
+
+	count := 0
+	err = meta.EachListItem(obj, func(obj runtime.Object) error {
+		count++
+		return setSelfLink(obj, requestInfo, namer)
+	})
+	return count, err
+}

pkg/apiserver/installer/apiserver_test.go

@@ -85,9 +85,9 @@ func init() {
 		&metav1.APIResourceList{},
 	)
 
-	customMetricsGroupInfo = genericapiserver.NewDefaultAPIGroupInfo(custom_metrics.GroupName, Scheme, metav1.ParameterCodec, Codecs)
+	customMetricsGroupInfo = genericapiserver.NewDefaultAPIGroupInfo(custom_metrics.GroupName, Scheme, runtime.NewParameterCodec(Scheme), Codecs)
 	customMetricsGroupVersion = customMetricsGroupInfo.PrioritizedVersions[0]
-	externalMetricsGroupInfo = genericapiserver.NewDefaultAPIGroupInfo(external_metrics.GroupName, Scheme, metav1.ParameterCodec, Codecs)
+	externalMetricsGroupInfo = genericapiserver.NewDefaultAPIGroupInfo(external_metrics.GroupName, Scheme, runtime.NewParameterCodec(Scheme), Codecs)
 	externalMetricsGroupVersion = externalMetricsGroupInfo.PrioritizedVersions[0]
 }
 
@@ -165,7 +165,7 @@ func handleExternalMetrics(prov provider.ExternalMetricsProvider) http.Handler {
 			Linker:          runtime.SelfLinker(meta.NewAccessor()),
 		},
 		ResourceLister: provider.NewExternalMetricResourceLister(prov),
-		Handlers:       &CMHandlers{},
+		Handlers:       &EMHandlers{},
 	}
 
 	if err := group.InstallREST(container); err != nil {
@@ -186,7 +186,7 @@ type fakeCMProvider struct {
 	metrics                []provider.CustomMetricInfo
 }
 
-func (p *fakeCMProvider) valuesFor(name types.NamespacedName, info provider.CustomMetricInfo) (string, []custom_metrics.MetricValue, bool) {
+func (p *fakeCMProvider) valuesFor(name types.NamespacedName, info provider.CustomMetricInfo, metricSelector labels.Selector) (string, []custom_metrics.MetricValue, bool) {
 	if info.Namespaced {
 		metricId := name.Namespace + "/" + info.GroupResource.String() + "/" + name.Name + "/" + info.Metric
 		values, ok := p.namespacedValues[metricId]
@@ -198,17 +198,17 @@ func (p *fakeCMProvider) valuesFor(name types.NamespacedName, info provider.Cust
 	}
 }
 
-func (p *fakeCMProvider) GetMetricByName(name types.NamespacedName, info provider.CustomMetricInfo) (*custom_metrics.MetricValue, error) {
-	metricId, values, ok := p.valuesFor(name, info)
+func (p *fakeCMProvider) GetMetricByName(name types.NamespacedName, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValue, error) {
+	metricId, values, ok := p.valuesFor(name, info, metricSelector)
 	if !ok {
 		return nil, fmt.Errorf("non-existent metric requested (id: %s)", metricId)
 	}
 
 	return &values[0], nil
 }
 
-func (p *fakeCMProvider) GetMetricBySelector(namespace string, selector labels.Selector, info provider.CustomMetricInfo) (*custom_metrics.MetricValueList, error) {
-	metricId, values, ok := p.valuesFor(types.NamespacedName{Namespace: namespace, Name: "*"}, info)
+func (p *fakeCMProvider) GetMetricBySelector(namespace string, selector labels.Selector, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValueList, error) {
+	metricId, values, ok := p.valuesFor(types.NamespacedName{Namespace: namespace, Name: "*"}, info, metricSelector)
 	if !ok {
 		return nil, fmt.Errorf("non-existent metric requested (id: %s)", metricId)
 	}