blob: 927403cb2329094026396a99abdb0e0fdaa8ed14 [file] [log] [blame]
Matthias Andreas Benkard832a54e2019-01-29 09:27:38 +01001/*
2Copyright 2014 The Kubernetes Authors.
3
4Licensed under the Apache License, Version 2.0 (the "License");
5you may not use this file except in compliance with the License.
6You may obtain a copy of the License at
7
8 http://www.apache.org/licenses/LICENSE-2.0
9
10Unless required by applicable law or agreed to in writing, software
11distributed under the License is distributed on an "AS IS" BASIS,
12WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13See the License for the specific language governing permissions and
14limitations under the License.
15*/
16
17package rest
18
19import (
20 "fmt"
21 "mime"
22 "net/http"
23 "net/url"
24 "os"
25 "strconv"
26 "strings"
27 "time"
28
29 "k8s.io/apimachinery/pkg/runtime"
30 "k8s.io/apimachinery/pkg/runtime/schema"
31 "k8s.io/apimachinery/pkg/types"
32 "k8s.io/client-go/util/flowcontrol"
33)
34
35const (
36 // Environment variables: Note that the duration should be long enough that the backoff
37 // persists for some reasonable time (i.e. 120 seconds). The typical base might be "1".
38 envBackoffBase = "KUBE_CLIENT_BACKOFF_BASE"
39 envBackoffDuration = "KUBE_CLIENT_BACKOFF_DURATION"
40)
41
42// Interface captures the set of operations for generically interacting with Kubernetes REST apis.
43type Interface interface {
44 GetRateLimiter() flowcontrol.RateLimiter
45 Verb(verb string) *Request
46 Post() *Request
47 Put() *Request
48 Patch(pt types.PatchType) *Request
49 Get() *Request
50 Delete() *Request
51 APIVersion() schema.GroupVersion
52}
53
54// RESTClient imposes common Kubernetes API conventions on a set of resource paths.
55// The baseURL is expected to point to an HTTP or HTTPS path that is the parent
56// of one or more resources. The server should return a decodable API resource
57// object, or an api.Status object which contains information about the reason for
58// any failure.
59//
60// Most consumers should use client.New() to get a Kubernetes API client.
61type RESTClient struct {
62 // base is the root URL for all invocations of the client
63 base *url.URL
64 // versionedAPIPath is a path segment connecting the base URL to the resource root
65 versionedAPIPath string
66
67 // contentConfig is the information used to communicate with the server.
68 contentConfig ContentConfig
69
70 // serializers contain all serializers for underlying content type.
71 serializers Serializers
72
73 // creates BackoffManager that is passed to requests.
74 createBackoffMgr func() BackoffManager
75
76 // TODO extract this into a wrapper interface via the RESTClient interface in kubectl.
77 Throttle flowcontrol.RateLimiter
78
79 // Set specific behavior of the client. If not set http.DefaultClient will be used.
80 Client *http.Client
81}
82
83type Serializers struct {
84 Encoder runtime.Encoder
85 Decoder runtime.Decoder
86 StreamingSerializer runtime.Serializer
87 Framer runtime.Framer
88 RenegotiatedDecoder func(contentType string, params map[string]string) (runtime.Decoder, error)
89}
90
91// NewRESTClient creates a new RESTClient. This client performs generic REST functions
92// such as Get, Put, Post, and Delete on specified paths. Codec controls encoding and
93// decoding of responses from the server.
94func NewRESTClient(baseURL *url.URL, versionedAPIPath string, config ContentConfig, maxQPS float32, maxBurst int, rateLimiter flowcontrol.RateLimiter, client *http.Client) (*RESTClient, error) {
95 base := *baseURL
96 if !strings.HasSuffix(base.Path, "/") {
97 base.Path += "/"
98 }
99 base.RawQuery = ""
100 base.Fragment = ""
101
102 if config.GroupVersion == nil {
103 config.GroupVersion = &schema.GroupVersion{}
104 }
105 if len(config.ContentType) == 0 {
106 config.ContentType = "application/json"
107 }
108 serializers, err := createSerializers(config)
109 if err != nil {
110 return nil, err
111 }
112
113 var throttle flowcontrol.RateLimiter
114 if maxQPS > 0 && rateLimiter == nil {
115 throttle = flowcontrol.NewTokenBucketRateLimiter(maxQPS, maxBurst)
116 } else if rateLimiter != nil {
117 throttle = rateLimiter
118 }
119 return &RESTClient{
120 base: &base,
121 versionedAPIPath: versionedAPIPath,
122 contentConfig: config,
123 serializers: *serializers,
124 createBackoffMgr: readExpBackoffConfig,
125 Throttle: throttle,
126 Client: client,
127 }, nil
128}
129
130// GetRateLimiter returns rate limier for a given client, or nil if it's called on a nil client
131func (c *RESTClient) GetRateLimiter() flowcontrol.RateLimiter {
132 if c == nil {
133 return nil
134 }
135 return c.Throttle
136}
137
138// readExpBackoffConfig handles the internal logic of determining what the
139// backoff policy is. By default if no information is available, NoBackoff.
140// TODO Generalize this see #17727 .
141func readExpBackoffConfig() BackoffManager {
142 backoffBase := os.Getenv(envBackoffBase)
143 backoffDuration := os.Getenv(envBackoffDuration)
144
145 backoffBaseInt, errBase := strconv.ParseInt(backoffBase, 10, 64)
146 backoffDurationInt, errDuration := strconv.ParseInt(backoffDuration, 10, 64)
147 if errBase != nil || errDuration != nil {
148 return &NoBackoff{}
149 }
150 return &URLBackoff{
151 Backoff: flowcontrol.NewBackOff(
152 time.Duration(backoffBaseInt)*time.Second,
153 time.Duration(backoffDurationInt)*time.Second)}
154}
155
156// createSerializers creates all necessary serializers for given contentType.
157// TODO: the negotiated serializer passed to this method should probably return
158// serializers that control decoding and versioning without this package
159// being aware of the types. Depends on whether RESTClient must deal with
160// generic infrastructure.
161func createSerializers(config ContentConfig) (*Serializers, error) {
162 mediaTypes := config.NegotiatedSerializer.SupportedMediaTypes()
163 contentType := config.ContentType
164 mediaType, _, err := mime.ParseMediaType(contentType)
165 if err != nil {
166 return nil, fmt.Errorf("the content type specified in the client configuration is not recognized: %v", err)
167 }
168 info, ok := runtime.SerializerInfoForMediaType(mediaTypes, mediaType)
169 if !ok {
170 if len(contentType) != 0 || len(mediaTypes) == 0 {
171 return nil, fmt.Errorf("no serializers registered for %s", contentType)
172 }
173 info = mediaTypes[0]
174 }
175
176 internalGV := schema.GroupVersions{
177 {
178 Group: config.GroupVersion.Group,
179 Version: runtime.APIVersionInternal,
180 },
181 // always include the legacy group as a decoding target to handle non-error `Status` return types
182 {
183 Group: "",
184 Version: runtime.APIVersionInternal,
185 },
186 }
187
188 s := &Serializers{
189 Encoder: config.NegotiatedSerializer.EncoderForVersion(info.Serializer, *config.GroupVersion),
190 Decoder: config.NegotiatedSerializer.DecoderToVersion(info.Serializer, internalGV),
191
192 RenegotiatedDecoder: func(contentType string, params map[string]string) (runtime.Decoder, error) {
193 info, ok := runtime.SerializerInfoForMediaType(mediaTypes, contentType)
194 if !ok {
195 return nil, fmt.Errorf("serializer for %s not registered", contentType)
196 }
197 return config.NegotiatedSerializer.DecoderToVersion(info.Serializer, internalGV), nil
198 },
199 }
200 if info.StreamSerializer != nil {
201 s.StreamingSerializer = info.StreamSerializer.Serializer
202 s.Framer = info.StreamSerializer.Framer
203 }
204
205 return s, nil
206}
207
208// Verb begins a request with a verb (GET, POST, PUT, DELETE).
209//
210// Example usage of RESTClient's request building interface:
211// c, err := NewRESTClient(...)
212// if err != nil { ... }
213// resp, err := c.Verb("GET").
214// Path("pods").
215// SelectorParam("labels", "area=staging").
216// Timeout(10*time.Second).
217// Do()
218// if err != nil { ... }
219// list, ok := resp.(*api.PodList)
220//
221func (c *RESTClient) Verb(verb string) *Request {
222 backoff := c.createBackoffMgr()
223
224 if c.Client == nil {
225 return NewRequest(nil, verb, c.base, c.versionedAPIPath, c.contentConfig, c.serializers, backoff, c.Throttle, 0)
226 }
227 return NewRequest(c.Client, verb, c.base, c.versionedAPIPath, c.contentConfig, c.serializers, backoff, c.Throttle, c.Client.Timeout)
228}
229
230// Post begins a POST request. Short for c.Verb("POST").
231func (c *RESTClient) Post() *Request {
232 return c.Verb("POST")
233}
234
235// Put begins a PUT request. Short for c.Verb("PUT").
236func (c *RESTClient) Put() *Request {
237 return c.Verb("PUT")
238}
239
240// Patch begins a PATCH request. Short for c.Verb("Patch").
241func (c *RESTClient) Patch(pt types.PatchType) *Request {
242 return c.Verb("PATCH").SetHeader("Content-Type", string(pt))
243}
244
245// Get begins a GET request. Short for c.Verb("GET").
246func (c *RESTClient) Get() *Request {
247 return c.Verb("GET")
248}
249
250// Delete begins a DELETE request. Short for c.Verb("DELETE").
251func (c *RESTClient) Delete() *Request {
252 return c.Verb("DELETE")
253}
254
255// APIVersion returns the APIVersion this RESTClient is expected to use.
256func (c *RESTClient) APIVersion() schema.GroupVersion {
257 return *c.contentConfig.GroupVersion
258}