Update go dependencies

This commit is contained in:
Manuel Alejandro de Brito Fontes 2020-03-24 10:44:24 -03:00
parent a46126a034
commit 3eafaa35a1
1108 changed files with 32555 additions and 83490 deletions

View file

@ -0,0 +1,97 @@
/*
Copyright 2018 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 apiutil contains utilities for working with raw Kubernetes
// API machinery, such as creating RESTMappers and raw REST clients,
// and extracting the GVK of an object.
package apiutil
import (
"fmt"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/client-go/discovery"
"k8s.io/client-go/rest"
"k8s.io/client-go/restmapper"
)
// NewDiscoveryRESTMapper constructs a new RESTMapper based on discovery
// information fetched by a new client with the given config.
func NewDiscoveryRESTMapper(c *rest.Config) (meta.RESTMapper, error) {
// Get a mapper
dc, err := discovery.NewDiscoveryClientForConfig(c)
if err != nil {
return nil, err
}
gr, err := restmapper.GetAPIGroupResources(dc)
if err != nil {
return nil, err
}
return restmapper.NewDiscoveryRESTMapper(gr), nil
}
// GVKForObject finds the GroupVersionKind associated with the given object, if there is only a single such GVK.
func GVKForObject(obj runtime.Object, scheme *runtime.Scheme) (schema.GroupVersionKind, error) {
gvks, isUnversioned, err := scheme.ObjectKinds(obj)
if err != nil {
return schema.GroupVersionKind{}, err
}
if isUnversioned {
return schema.GroupVersionKind{}, fmt.Errorf("cannot create a new informer for the unversioned type %T", obj)
}
if len(gvks) < 1 {
return schema.GroupVersionKind{}, fmt.Errorf("no group-version-kinds associated with type %T", obj)
}
if len(gvks) > 1 {
// this should only trigger for things like metav1.XYZ --
// normal versioned types should be fine
return schema.GroupVersionKind{}, fmt.Errorf(
"multiple group-version-kinds associated with type %T, refusing to guess at one", obj)
}
return gvks[0], nil
}
// RESTClientForGVK constructs a new rest.Interface capable of accessing the resource associated
// with the given GroupVersionKind. The REST client will be configured to use the negotiated serializer from
// baseConfig, if set, otherwise a default serializer will be set.
func RESTClientForGVK(gvk schema.GroupVersionKind, baseConfig *rest.Config, codecs serializer.CodecFactory) (rest.Interface, error) {
cfg := createRestConfig(gvk, baseConfig)
if cfg.NegotiatedSerializer == nil {
cfg.NegotiatedSerializer = serializer.WithoutConversionCodecFactory{CodecFactory: codecs}
}
return rest.RESTClientFor(cfg)
}
//createRestConfig copies the base config and updates needed fields for a new rest config
func createRestConfig(gvk schema.GroupVersionKind, baseConfig *rest.Config) *rest.Config {
gv := gvk.GroupVersion()
cfg := rest.CopyConfig(baseConfig)
cfg.GroupVersion = &gv
if gvk.Group == "" {
cfg.APIPath = "/api"
} else {
cfg.APIPath = "/apis"
}
if cfg.UserAgent == "" {
cfg.UserAgent = rest.DefaultKubernetesUserAgent()
}
return cfg
}

View file

@ -0,0 +1,323 @@
/*
Copyright 2019 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 apiutil
import (
"errors"
"sync"
"time"
"golang.org/x/time/rate"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/discovery"
"k8s.io/client-go/rest"
"k8s.io/client-go/restmapper"
)
// ErrRateLimited is returned by a RESTMapper method if the number of API
// calls has exceeded a limit within a certain time period.
type ErrRateLimited struct {
// Duration to wait until the next API call can be made.
Delay time.Duration
}
func (e ErrRateLimited) Error() string {
return "too many API calls to the RESTMapper within a timeframe"
}
// DelayIfRateLimited returns the delay time until the next API call is
// allowed and true if err is of type ErrRateLimited. The zero
// time.Duration value and false are returned if err is not a ErrRateLimited.
func DelayIfRateLimited(err error) (time.Duration, bool) {
var rlerr ErrRateLimited
if errors.As(err, &rlerr) {
return rlerr.Delay, true
}
return 0, false
}
// dynamicRESTMapper is a RESTMapper that dynamically discovers resource
// types at runtime.
type dynamicRESTMapper struct {
mu sync.RWMutex // protects the following fields
staticMapper meta.RESTMapper
limiter *dynamicLimiter
newMapper func() (meta.RESTMapper, error)
lazy bool
// Used for lazy init.
initOnce sync.Once
}
// DynamicRESTMapperOption is a functional option on the dynamicRESTMapper
type DynamicRESTMapperOption func(*dynamicRESTMapper) error
// WithLimiter sets the RESTMapper's underlying limiter to lim.
func WithLimiter(lim *rate.Limiter) DynamicRESTMapperOption {
return func(drm *dynamicRESTMapper) error {
drm.limiter = &dynamicLimiter{lim}
return nil
}
}
// WithLazyDiscovery prevents the RESTMapper from discovering REST mappings
// until an API call is made.
var WithLazyDiscovery DynamicRESTMapperOption = func(drm *dynamicRESTMapper) error {
drm.lazy = true
return nil
}
// WithCustomMapper supports setting a custom RESTMapper refresher instead of
// the default method, which uses a discovery client.
//
// This exists mainly for testing, but can be useful if you need tighter control
// over how discovery is performed, which discovery endpoints are queried, etc.
func WithCustomMapper(newMapper func() (meta.RESTMapper, error)) DynamicRESTMapperOption {
return func(drm *dynamicRESTMapper) error {
drm.newMapper = newMapper
return nil
}
}
// NewDynamicRESTMapper returns a dynamic RESTMapper for cfg. The dynamic
// RESTMapper dynamically discovers resource types at runtime. opts
// configure the RESTMapper.
func NewDynamicRESTMapper(cfg *rest.Config, opts ...DynamicRESTMapperOption) (meta.RESTMapper, error) {
client, err := discovery.NewDiscoveryClientForConfig(cfg)
if err != nil {
return nil, err
}
drm := &dynamicRESTMapper{
limiter: &dynamicLimiter{
rate.NewLimiter(rate.Limit(defaultRefillRate), defaultLimitSize),
},
newMapper: func() (meta.RESTMapper, error) {
groupResources, err := restmapper.GetAPIGroupResources(client)
if err != nil {
return nil, err
}
return restmapper.NewDiscoveryRESTMapper(groupResources), nil
},
}
for _, opt := range opts {
if err = opt(drm); err != nil {
return nil, err
}
}
if !drm.lazy {
if err := drm.setStaticMapper(); err != nil {
return nil, err
}
}
return drm, nil
}
var (
// defaultRefilRate is the default rate at which potential calls are
// added back to the "bucket" of allowed calls.
defaultRefillRate = 5
// defaultLimitSize is the default starting/max number of potential calls
// per second. Once a call is used, it's added back to the bucket at a rate
// of defaultRefillRate per second.
defaultLimitSize = 5
)
// setStaticMapper sets drm's staticMapper by querying its client, regardless
// of reload backoff.
func (drm *dynamicRESTMapper) setStaticMapper() error {
newMapper, err := drm.newMapper()
if err != nil {
return err
}
drm.staticMapper = newMapper
return nil
}
// init initializes drm only once if drm is lazy.
func (drm *dynamicRESTMapper) init() (err error) {
drm.initOnce.Do(func() {
if drm.lazy {
err = drm.setStaticMapper()
}
})
return err
}
// checkAndReload attempts to call the given callback, which is assumed to be dependent
// on the data in the restmapper.
//
// If the callback returns a NoKindMatchError, it will attempt to reload
// the RESTMapper's data and re-call the callback once that's occurred.
// If the callback returns any other error, the function will return immediately regardless.
//
// It will take care
// ensuring that reloads are rate-limitted and that extraneous calls aren't made.
// It's thread-safe, and worries about thread-safety for the callback (so the callback does
// not need to attempt to lock the restmapper).
func (drm *dynamicRESTMapper) checkAndReload(needsReloadErr error, checkNeedsReload func() error) error {
// first, check the common path -- data is fresh enough
// (use an IIFE for the lock's defer)
err := func() error {
drm.mu.RLock()
defer drm.mu.RUnlock()
return checkNeedsReload()
}()
// NB(directxman12): `Is` and `As` have a confusing relationship --
// `Is` is like `== or does this implement .Is`, whereas `As` says
// `can I type-assert into`
needsReload := errors.As(err, &needsReloadErr)
if !needsReload {
return err
}
// if the data wasn't fresh, we'll need to try and update it, so grab the lock...
drm.mu.Lock()
defer drm.mu.Unlock()
// ... and double-check that we didn't reload in the meantime
err = checkNeedsReload()
needsReload = errors.As(err, &needsReloadErr)
if !needsReload {
return err
}
// we're still stale, so grab a rate-limit token if we can...
if err := drm.limiter.checkRate(); err != nil {
return err
}
// ...reload...
if err := drm.setStaticMapper(); err != nil {
return err
}
// ...and return the results of the closure regardless
return checkNeedsReload()
}
// TODO: wrap reload errors on NoKindMatchError with go 1.13 errors.
func (drm *dynamicRESTMapper) KindFor(resource schema.GroupVersionResource) (schema.GroupVersionKind, error) {
if err := drm.init(); err != nil {
return schema.GroupVersionKind{}, err
}
var gvk schema.GroupVersionKind
err := drm.checkAndReload(&meta.NoResourceMatchError{}, func() error {
var err error
gvk, err = drm.staticMapper.KindFor(resource)
return err
})
return gvk, err
}
func (drm *dynamicRESTMapper) KindsFor(resource schema.GroupVersionResource) ([]schema.GroupVersionKind, error) {
if err := drm.init(); err != nil {
return nil, err
}
var gvks []schema.GroupVersionKind
err := drm.checkAndReload(&meta.NoResourceMatchError{}, func() error {
var err error
gvks, err = drm.staticMapper.KindsFor(resource)
return err
})
return gvks, err
}
func (drm *dynamicRESTMapper) ResourceFor(input schema.GroupVersionResource) (schema.GroupVersionResource, error) {
if err := drm.init(); err != nil {
return schema.GroupVersionResource{}, err
}
var gvr schema.GroupVersionResource
err := drm.checkAndReload(&meta.NoResourceMatchError{}, func() error {
var err error
gvr, err = drm.staticMapper.ResourceFor(input)
return err
})
return gvr, err
}
func (drm *dynamicRESTMapper) ResourcesFor(input schema.GroupVersionResource) ([]schema.GroupVersionResource, error) {
if err := drm.init(); err != nil {
return nil, err
}
var gvrs []schema.GroupVersionResource
err := drm.checkAndReload(&meta.NoResourceMatchError{}, func() error {
var err error
gvrs, err = drm.staticMapper.ResourcesFor(input)
return err
})
return gvrs, err
}
func (drm *dynamicRESTMapper) RESTMapping(gk schema.GroupKind, versions ...string) (*meta.RESTMapping, error) {
if err := drm.init(); err != nil {
return nil, err
}
var mapping *meta.RESTMapping
err := drm.checkAndReload(&meta.NoKindMatchError{}, func() error {
var err error
mapping, err = drm.staticMapper.RESTMapping(gk, versions...)
return err
})
return mapping, err
}
func (drm *dynamicRESTMapper) RESTMappings(gk schema.GroupKind, versions ...string) ([]*meta.RESTMapping, error) {
if err := drm.init(); err != nil {
return nil, err
}
var mappings []*meta.RESTMapping
err := drm.checkAndReload(&meta.NoKindMatchError{}, func() error {
var err error
mappings, err = drm.staticMapper.RESTMappings(gk, versions...)
return err
})
return mappings, err
}
func (drm *dynamicRESTMapper) ResourceSingularizer(resource string) (string, error) {
if err := drm.init(); err != nil {
return "", err
}
var singular string
err := drm.checkAndReload(&meta.NoResourceMatchError{}, func() error {
var err error
singular, err = drm.staticMapper.ResourceSingularizer(resource)
return err
})
return singular, err
}
// dynamicLimiter holds a rate limiter used to throttle chatty RESTMapper users.
type dynamicLimiter struct {
*rate.Limiter
}
// checkRate returns an ErrRateLimited if too many API calls have been made
// within the set limit.
func (b *dynamicLimiter) checkRate() error {
res := b.Reserve()
if res.Delay() == 0 {
return nil
}
res.Cancel()
return ErrRateLimited{res.Delay()}
}

View file

@ -0,0 +1,208 @@
/*
Copyright 2018 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 client
import (
"context"
"fmt"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
)
// Options are creation options for a Client
type Options struct {
// Scheme, if provided, will be used to map go structs to GroupVersionKinds
Scheme *runtime.Scheme
// Mapper, if provided, will be used to map GroupVersionKinds to Resources
Mapper meta.RESTMapper
}
// New returns a new Client using the provided config and Options.
// The returned client reads *and* writes directly from the server
// (it doesn't use object caches). It understands how to work with
// normal types (both custom resources and aggregated/built-in resources),
// as well as unstructured types.
//
// In the case of normal types, the scheme will be used to look up the
// corresponding group, version, and kind for the given type. In the
// case of unstructured types, the group, version, and kind will be extracted
// from the corresponding fields on the object.
func New(config *rest.Config, options Options) (Client, error) {
if config == nil {
return nil, fmt.Errorf("must provide non-nil rest.Config to client.New")
}
// Init a scheme if none provided
if options.Scheme == nil {
options.Scheme = scheme.Scheme
}
// Init a Mapper if none provided
if options.Mapper == nil {
var err error
options.Mapper, err = apiutil.NewDynamicRESTMapper(config)
if err != nil {
return nil, err
}
}
clientcache := &clientCache{
config: config,
scheme: options.Scheme,
mapper: options.Mapper,
codecs: serializer.NewCodecFactory(options.Scheme),
resourceByType: make(map[schema.GroupVersionKind]*resourceMeta),
}
c := &client{
typedClient: typedClient{
cache: clientcache,
paramCodec: runtime.NewParameterCodec(options.Scheme),
},
unstructuredClient: unstructuredClient{
cache: clientcache,
paramCodec: noConversionParamCodec{},
},
}
return c, nil
}
var _ Client = &client{}
// client is a client.Client that reads and writes directly from/to an API server. It lazily initializes
// new clients at the time they are used, and caches the client.
type client struct {
typedClient typedClient
unstructuredClient unstructuredClient
}
// resetGroupVersionKind is a helper function to restore and preserve GroupVersionKind on an object.
// TODO(vincepri): Remove this function and its calls once controller-runtime dependencies are upgraded to 1.16?
func (c *client) resetGroupVersionKind(obj runtime.Object, gvk schema.GroupVersionKind) {
if gvk != schema.EmptyObjectKind.GroupVersionKind() {
if v, ok := obj.(schema.ObjectKind); ok {
v.SetGroupVersionKind(gvk)
}
}
}
// Create implements client.Client
func (c *client) Create(ctx context.Context, obj runtime.Object, opts ...CreateOption) error {
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.Create(ctx, obj, opts...)
}
return c.typedClient.Create(ctx, obj, opts...)
}
// Update implements client.Client
func (c *client) Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
defer c.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.Update(ctx, obj, opts...)
}
return c.typedClient.Update(ctx, obj, opts...)
}
// Delete implements client.Client
func (c *client) Delete(ctx context.Context, obj runtime.Object, opts ...DeleteOption) error {
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.Delete(ctx, obj, opts...)
}
return c.typedClient.Delete(ctx, obj, opts...)
}
// DeleteAllOf implements client.Client
func (c *client) DeleteAllOf(ctx context.Context, obj runtime.Object, opts ...DeleteAllOfOption) error {
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.DeleteAllOf(ctx, obj, opts...)
}
return c.typedClient.DeleteAllOf(ctx, obj, opts...)
}
// Patch implements client.Client
func (c *client) Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
defer c.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.Patch(ctx, obj, patch, opts...)
}
return c.typedClient.Patch(ctx, obj, patch, opts...)
}
// Get implements client.Client
func (c *client) Get(ctx context.Context, key ObjectKey, obj runtime.Object) error {
_, ok := obj.(*unstructured.Unstructured)
if ok {
return c.unstructuredClient.Get(ctx, key, obj)
}
return c.typedClient.Get(ctx, key, obj)
}
// List implements client.Client
func (c *client) List(ctx context.Context, obj runtime.Object, opts ...ListOption) error {
_, ok := obj.(*unstructured.UnstructuredList)
if ok {
return c.unstructuredClient.List(ctx, obj, opts...)
}
return c.typedClient.List(ctx, obj, opts...)
}
// Status implements client.StatusClient
func (c *client) Status() StatusWriter {
return &statusWriter{client: c}
}
// statusWriter is client.StatusWriter that writes status subresource
type statusWriter struct {
client *client
}
// ensure statusWriter implements client.StatusWriter
var _ StatusWriter = &statusWriter{}
// Update implements client.StatusWriter
func (sw *statusWriter) Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
defer sw.client.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
_, ok := obj.(*unstructured.Unstructured)
if ok {
return sw.client.unstructuredClient.UpdateStatus(ctx, obj, opts...)
}
return sw.client.typedClient.UpdateStatus(ctx, obj, opts...)
}
// Patch implements client.Client
func (sw *statusWriter) Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
defer sw.client.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
_, ok := obj.(*unstructured.Unstructured)
if ok {
return sw.client.unstructuredClient.PatchStatus(ctx, obj, patch, opts...)
}
return sw.client.typedClient.PatchStatus(ctx, obj, patch, opts...)
}

View file

@ -0,0 +1,140 @@
/*
Copyright 2018 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 client
import (
"strings"
"sync"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
)
// clientCache creates and caches rest clients and metadata for Kubernetes types
type clientCache struct {
// config is the rest.Config to talk to an apiserver
config *rest.Config
// scheme maps go structs to GroupVersionKinds
scheme *runtime.Scheme
// mapper maps GroupVersionKinds to Resources
mapper meta.RESTMapper
// codecs are used to create a REST client for a gvk
codecs serializer.CodecFactory
// resourceByType caches type metadata
resourceByType map[schema.GroupVersionKind]*resourceMeta
mu sync.RWMutex
}
// newResource maps obj to a Kubernetes Resource and constructs a client for that Resource.
// If the object is a list, the resource represents the item's type instead.
func (c *clientCache) newResource(gvk schema.GroupVersionKind, isList bool) (*resourceMeta, error) {
if strings.HasSuffix(gvk.Kind, "List") && isList {
// if this was a list, treat it as a request for the item's resource
gvk.Kind = gvk.Kind[:len(gvk.Kind)-4]
}
client, err := apiutil.RESTClientForGVK(gvk, c.config, c.codecs)
if err != nil {
return nil, err
}
mapping, err := c.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
if err != nil {
return nil, err
}
return &resourceMeta{Interface: client, mapping: mapping, gvk: gvk}, nil
}
// getResource returns the resource meta information for the given type of object.
// If the object is a list, the resource represents the item's type instead.
func (c *clientCache) getResource(obj runtime.Object) (*resourceMeta, error) {
gvk, err := apiutil.GVKForObject(obj, c.scheme)
if err != nil {
return nil, err
}
// It's better to do creation work twice than to not let multiple
// people make requests at once
c.mu.RLock()
r, known := c.resourceByType[gvk]
c.mu.RUnlock()
if known {
return r, nil
}
// Initialize a new Client
c.mu.Lock()
defer c.mu.Unlock()
r, err = c.newResource(gvk, meta.IsListType(obj))
if err != nil {
return nil, err
}
c.resourceByType[gvk] = r
return r, err
}
// getObjMeta returns objMeta containing both type and object metadata and state
func (c *clientCache) getObjMeta(obj runtime.Object) (*objMeta, error) {
r, err := c.getResource(obj)
if err != nil {
return nil, err
}
m, err := meta.Accessor(obj)
if err != nil {
return nil, err
}
return &objMeta{resourceMeta: r, Object: m}, err
}
// resourceMeta caches state for a Kubernetes type.
type resourceMeta struct {
// client is the rest client used to talk to the apiserver
rest.Interface
// gvk is the GroupVersionKind of the resourceMeta
gvk schema.GroupVersionKind
// mapping is the rest mapping
mapping *meta.RESTMapping
}
// isNamespaced returns true if the type is namespaced
func (r *resourceMeta) isNamespaced() bool {
return r.mapping.Scope.Name() != meta.RESTScopeNameRoot
}
// resource returns the resource name of the type
func (r *resourceMeta) resource() string {
return r.mapping.Resource.Resource
}
// objMeta stores type and object information about a Kubernetes type
type objMeta struct {
// resourceMeta contains type information for the object
*resourceMeta
// Object contains meta data for the object instance
metav1.Object
}

View file

@ -0,0 +1,24 @@
package client
import (
"errors"
"net/url"
"k8s.io/apimachinery/pkg/conversion/queryparams"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
)
var _ runtime.ParameterCodec = noConversionParamCodec{}
// noConversionParamCodec is a no-conversion codec for serializing parameters into URL query strings.
// it's useful in scenarios with the unstructured client and arbitrary resouces.
type noConversionParamCodec struct{}
func (noConversionParamCodec) EncodeParameters(obj runtime.Object, to schema.GroupVersion) (url.Values, error) {
return queryparams.Convert(obj)
}
func (noConversionParamCodec) DecodeParameters(parameters url.Values, from schema.GroupVersion, into runtime.Object) error {
return errors.New("DecodeParameters not implemented on noConversionParamCodec")
}

View file

@ -20,6 +20,8 @@ import (
"flag"
"fmt"
"os"
"os/user"
"path"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
@ -117,7 +119,21 @@ func loadConfig(context string) (*rest.Config, error) {
// If the recommended kubeconfig env variable is set, or there
// is no in-cluster config, try the default recommended locations.
if c, err := loadConfigWithContext(apiServerURL, clientcmd.NewDefaultClientConfigLoadingRules(), context); err == nil {
//
// NOTE: For default config file locations, upstream only checks
// $HOME for the user's home directory, but we can also try
// os/user.HomeDir when $HOME is unset.
//
// TODO(jlanford): could this be done upstream?
loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
if _, ok := os.LookupEnv("HOME"); !ok {
u, err := user.Current()
if err != nil {
return nil, fmt.Errorf("could not get current user: %v", err)
}
loadingRules.Precedence = append(loadingRules.Precedence, path.Join(u.HomeDir, clientcmd.RecommendedHomeDir, clientcmd.RecommendedFileName))
}
if c, err := loadConfigWithContext(apiServerURL, loadingRules, context); err == nil {
return c, nil
}

View file

@ -0,0 +1,49 @@
/*
Copyright 2018 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 client contains functionality for interacting with Kubernetes API
// servers.
//
// Clients
//
// Clients are split into two interfaces -- Readers and Writers. Readers
// get and list, while writers create, update, and delete.
//
// The New function can be used to create a new client that talks directly
// to the API server.
//
// A common pattern in Kubernetes to read from a cache and write to the API
// server. This pattern is covered by the DelegatingClient type, which can
// be used to have a client whose Reader is different from the Writer.
//
// Options
//
// Many client operations in Kubernetes support options. These options are
// represented as variadic arguments at the end of a given method call.
// For instance, to use a label selector on list, you can call
// err := someReader.List(context.Background(), &podList, client.MatchingLabels{"somelabel": "someval"})
//
// Indexing
//
// Indexes may be added to caches using a FieldIndexer. This allows you to easily
// and efficiently look up objects with certain properties. You can then make
// use of the index by specifying a field selector on calls to List on the Reader
// corresponding to the given Cache.
//
// For instance, a Secret controller might have an index on the
// `.spec.volumes.secret.secretName` field in Pod objects, so that it could
// easily look up all pods that reference a given secret.
package client

View file

@ -0,0 +1,135 @@
/*
Copyright 2018 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 client
import (
"context"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
)
// ObjectKey identifies a Kubernetes Object.
type ObjectKey = types.NamespacedName
// ObjectKeyFromObject returns the ObjectKey given a runtime.Object
func ObjectKeyFromObject(obj runtime.Object) (ObjectKey, error) {
accessor, err := meta.Accessor(obj)
if err != nil {
return ObjectKey{}, err
}
return ObjectKey{Namespace: accessor.GetNamespace(), Name: accessor.GetName()}, nil
}
// Patch is a patch that can be applied to a Kubernetes object.
type Patch interface {
// Type is the PatchType of the patch.
Type() types.PatchType
// Data is the raw data representing the patch.
Data(obj runtime.Object) ([]byte, error)
}
// TODO(directxman12): is there a sane way to deal with get/delete options?
// Reader knows how to read and list Kubernetes objects.
type Reader interface {
// Get retrieves an obj for the given object key from the Kubernetes Cluster.
// obj must be a struct pointer so that obj can be updated with the response
// returned by the Server.
Get(ctx context.Context, key ObjectKey, obj runtime.Object) error
// List retrieves list of objects for a given namespace and list options. On a
// successful call, Items field in the list will be populated with the
// result returned from the server.
List(ctx context.Context, list runtime.Object, opts ...ListOption) error
}
// Writer knows how to create, delete, and update Kubernetes objects.
type Writer interface {
// Create saves the object obj in the Kubernetes cluster.
Create(ctx context.Context, obj runtime.Object, opts ...CreateOption) error
// Delete deletes the given obj from Kubernetes cluster.
Delete(ctx context.Context, obj runtime.Object, opts ...DeleteOption) error
// Update updates the given obj in the Kubernetes cluster. obj must be a
// struct pointer so that obj can be updated with the content returned by the Server.
Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error
// Patch patches the given obj in the Kubernetes cluster. obj must be a
// struct pointer so that obj can be updated with the content returned by the Server.
Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error
// DeleteAllOf deletes all objects of the given type matching the given options.
DeleteAllOf(ctx context.Context, obj runtime.Object, opts ...DeleteAllOfOption) error
}
// StatusClient knows how to create a client which can update status subresource
// for kubernetes objects.
type StatusClient interface {
Status() StatusWriter
}
// StatusWriter knows how to update status subresource of a Kubernetes object.
type StatusWriter interface {
// Update updates the fields corresponding to the status subresource for the
// given obj. obj must be a struct pointer so that obj can be updated
// with the content returned by the Server.
Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error
// Patch patches the given object's subresource. obj must be a struct
// pointer so that obj can be updated with the content returned by the
// Server.
Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error
}
// Client knows how to perform CRUD operations on Kubernetes objects.
type Client interface {
Reader
Writer
StatusClient
}
// IndexerFunc knows how to take an object and turn it into a series
// of non-namespaced keys. Namespaced objects are automatically given
// namespaced and non-spaced variants, so keys do not need to include namespace.
type IndexerFunc func(runtime.Object) []string
// FieldIndexer knows how to index over a particular "field" such that it
// can later be used by a field selector.
type FieldIndexer interface {
// IndexFields adds an index with the given field name on the given object type
// by using the given function to extract the value for that field. If you want
// compatibility with the Kubernetes API server, only return one key, and only use
// fields that the API server supports. Otherwise, you can return multiple keys,
// and "equality" in the field selector means that at least one key matches the value.
// The FieldIndexer will automatically take care of indexing over namespace
// and supporting efficient all-namespace queries.
IndexField(ctx context.Context, obj runtime.Object, field string, extractValue IndexerFunc) error
}
// IgnoreNotFound returns nil on NotFound errors.
// All other values that are not NotFound errors or nil are returned unmodified.
func IgnoreNotFound(err error) error {
if apierrors.IsNotFound(err) {
return nil
}
return err
}

View file

@ -0,0 +1,675 @@
/*
Copyright 2018 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 client
import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
)
// {{{ "Functional" Option Interfaces
// CreateOption is some configuration that modifies options for a create request.
type CreateOption interface {
// ApplyToCreate applies this configuration to the given create options.
ApplyToCreate(*CreateOptions)
}
// DeleteOption is some configuration that modifies options for a delete request.
type DeleteOption interface {
// ApplyToDelete applies this configuration to the given delete options.
ApplyToDelete(*DeleteOptions)
}
// ListOption is some configuration that modifies options for a list request.
type ListOption interface {
// ApplyToList applies this configuration to the given list options.
ApplyToList(*ListOptions)
}
// UpdateOption is some configuration that modifies options for a update request.
type UpdateOption interface {
// ApplyToUpdate applies this configuration to the given update options.
ApplyToUpdate(*UpdateOptions)
}
// PatchOption is some configuration that modifies options for a patch request.
type PatchOption interface {
// ApplyToPatch applies this configuration to the given patch options.
ApplyToPatch(*PatchOptions)
}
// DeleteAllOfOption is some configuration that modifies options for a delete request.
type DeleteAllOfOption interface {
// ApplyToDeleteAllOf applies this configuration to the given deletecollection options.
ApplyToDeleteAllOf(*DeleteAllOfOptions)
}
// }}}
// {{{ Multi-Type Options
// DryRunAll sets the "dry run" option to "all", executing all
// validation, etc without persisting the change to storage.
var DryRunAll = dryRunAll{}
type dryRunAll struct{}
func (dryRunAll) ApplyToCreate(opts *CreateOptions) {
opts.DryRun = []string{metav1.DryRunAll}
}
func (dryRunAll) ApplyToUpdate(opts *UpdateOptions) {
opts.DryRun = []string{metav1.DryRunAll}
}
func (dryRunAll) ApplyToPatch(opts *PatchOptions) {
opts.DryRun = []string{metav1.DryRunAll}
}
func (dryRunAll) ApplyToDelete(opts *DeleteOptions) {
opts.DryRun = []string{metav1.DryRunAll}
}
// FieldOwner set the field manager name for the given server-side apply patch.
type FieldOwner string
func (f FieldOwner) ApplyToPatch(opts *PatchOptions) {
opts.FieldManager = string(f)
}
func (f FieldOwner) ApplyToCreate(opts *CreateOptions) {
opts.FieldManager = string(f)
}
func (f FieldOwner) ApplyToUpdate(opts *UpdateOptions) {
opts.FieldManager = string(f)
}
// }}}
// {{{ Create Options
// CreateOptions contains options for create requests. It's generally a subset
// of metav1.CreateOptions.
type CreateOptions struct {
// When present, indicates that modifications should not be
// persisted. An invalid or unrecognized dryRun directive will
// result in an error response and no further processing of the
// request. Valid values are:
// - All: all dry run stages will be processed
DryRun []string
// FieldManager is the name of the user or component submitting
// this request. It must be set with server-side apply.
FieldManager string
// Raw represents raw CreateOptions, as passed to the API server.
Raw *metav1.CreateOptions
}
// AsCreateOptions returns these options as a metav1.CreateOptions.
// This may mutate the Raw field.
func (o *CreateOptions) AsCreateOptions() *metav1.CreateOptions {
if o == nil {
return &metav1.CreateOptions{}
}
if o.Raw == nil {
o.Raw = &metav1.CreateOptions{}
}
o.Raw.DryRun = o.DryRun
o.Raw.FieldManager = o.FieldManager
return o.Raw
}
// ApplyOptions applies the given create options on these options,
// and then returns itself (for convenient chaining).
func (o *CreateOptions) ApplyOptions(opts []CreateOption) *CreateOptions {
for _, opt := range opts {
opt.ApplyToCreate(o)
}
return o
}
// ApplyToCreate implements CreateOption
func (o *CreateOptions) ApplyToCreate(co *CreateOptions) {
if o.DryRun != nil {
co.DryRun = o.DryRun
}
if o.FieldManager != "" {
co.FieldManager = o.FieldManager
}
if o.Raw != nil {
co.Raw = o.Raw
}
}
var _ CreateOption = &CreateOptions{}
// CreateDryRunAll sets the "dry run" option to "all".
//
// Deprecated: Use DryRunAll
var CreateDryRunAll = DryRunAll
// }}}
// {{{ Delete Options
// DeleteOptions contains options for delete requests. It's generally a subset
// of metav1.DeleteOptions.
type DeleteOptions struct {
// GracePeriodSeconds is the duration in seconds before the object should be
// deleted. Value must be non-negative integer. The value zero indicates
// delete immediately. If this value is nil, the default grace period for the
// specified type will be used.
GracePeriodSeconds *int64
// Preconditions must be fulfilled before a deletion is carried out. If not
// possible, a 409 Conflict status will be returned.
Preconditions *metav1.Preconditions
// PropagationPolicy determined whether and how garbage collection will be
// performed. Either this field or OrphanDependents may be set, but not both.
// The default policy is decided by the existing finalizer set in the
// metadata.finalizers and the resource-specific default policy.
// Acceptable values are: 'Orphan' - orphan the dependents; 'Background' -
// allow the garbage collector to delete the dependents in the background;
// 'Foreground' - a cascading policy that deletes all dependents in the
// foreground.
PropagationPolicy *metav1.DeletionPropagation
// Raw represents raw DeleteOptions, as passed to the API server.
Raw *metav1.DeleteOptions
// When present, indicates that modifications should not be
// persisted. An invalid or unrecognized dryRun directive will
// result in an error response and no further processing of the
// request. Valid values are:
// - All: all dry run stages will be processed
DryRun []string
}
// AsDeleteOptions returns these options as a metav1.DeleteOptions.
// This may mutate the Raw field.
func (o *DeleteOptions) AsDeleteOptions() *metav1.DeleteOptions {
if o == nil {
return &metav1.DeleteOptions{}
}
if o.Raw == nil {
o.Raw = &metav1.DeleteOptions{}
}
o.Raw.GracePeriodSeconds = o.GracePeriodSeconds
o.Raw.Preconditions = o.Preconditions
o.Raw.PropagationPolicy = o.PropagationPolicy
o.Raw.DryRun = o.DryRun
return o.Raw
}
// ApplyOptions applies the given delete options on these options,
// and then returns itself (for convenient chaining).
func (o *DeleteOptions) ApplyOptions(opts []DeleteOption) *DeleteOptions {
for _, opt := range opts {
opt.ApplyToDelete(o)
}
return o
}
var _ DeleteOption = &DeleteOptions{}
// ApplyToDelete implements DeleteOption
func (o *DeleteOptions) ApplyToDelete(do *DeleteOptions) {
if o.GracePeriodSeconds != nil {
do.GracePeriodSeconds = o.GracePeriodSeconds
}
if o.Preconditions != nil {
do.Preconditions = o.Preconditions
}
if o.PropagationPolicy != nil {
do.PropagationPolicy = o.PropagationPolicy
}
if o.Raw != nil {
do.Raw = o.Raw
}
if o.DryRun != nil {
do.DryRun = o.DryRun
}
}
// GracePeriodSeconds sets the grace period for the deletion
// to the given number of seconds.
type GracePeriodSeconds int64
func (s GracePeriodSeconds) ApplyToDelete(opts *DeleteOptions) {
secs := int64(s)
opts.GracePeriodSeconds = &secs
}
func (s GracePeriodSeconds) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
s.ApplyToDelete(&opts.DeleteOptions)
}
type Preconditions metav1.Preconditions
func (p Preconditions) ApplyToDelete(opts *DeleteOptions) {
preconds := metav1.Preconditions(p)
opts.Preconditions = &preconds
}
func (p Preconditions) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
p.ApplyToDelete(&opts.DeleteOptions)
}
type PropagationPolicy metav1.DeletionPropagation
func (p PropagationPolicy) ApplyToDelete(opts *DeleteOptions) {
policy := metav1.DeletionPropagation(p)
opts.PropagationPolicy = &policy
}
func (p PropagationPolicy) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
p.ApplyToDelete(&opts.DeleteOptions)
}
// }}}
// {{{ List Options
// ListOptions contains options for limiting or filtering results.
// It's generally a subset of metav1.ListOptions, with support for
// pre-parsed selectors (since generally, selectors will be executed
// against the cache).
type ListOptions struct {
// LabelSelector filters results by label. Use SetLabelSelector to
// set from raw string form.
LabelSelector labels.Selector
// FieldSelector filters results by a particular field. In order
// to use this with cache-based implementations, restrict usage to
// a single field-value pair that's been added to the indexers.
FieldSelector fields.Selector
// Namespace represents the namespace to list for, or empty for
// non-namespaced objects, or to list across all namespaces.
Namespace string
// Limit specifies the maximum number of results to return from the server. The server may
// not support this field on all resource types, but if it does and more results remain it
// will set the continue field on the returned list object. This field is not supported if watch
// is true in the Raw ListOptions.
Limit int64
// Continue is a token returned by the server that lets a client retrieve chunks of results
// from the server by specifying limit. The server may reject requests for continuation tokens
// it does not recognize and will return a 410 error if the token can no longer be used because
// it has expired. This field is not supported if watch is true in the Raw ListOptions.
Continue string
// Raw represents raw ListOptions, as passed to the API server. Note
// that these may not be respected by all implementations of interface,
// and the LabelSelector, FieldSelector, Limit and Continue fields are ignored.
Raw *metav1.ListOptions
}
var _ ListOption = &ListOptions{}
// ApplyToList implements ListOption for ListOptions
func (o *ListOptions) ApplyToList(lo *ListOptions) {
if o.LabelSelector != nil {
lo.LabelSelector = o.LabelSelector
}
if o.FieldSelector != nil {
lo.FieldSelector = o.FieldSelector
}
if o.Namespace != "" {
lo.Namespace = o.Namespace
}
if o.Raw != nil {
lo.Raw = o.Raw
}
if o.Limit > 0 {
lo.Limit = o.Limit
}
if o.Continue != "" {
lo.Continue = o.Continue
}
}
// AsListOptions returns these options as a flattened metav1.ListOptions.
// This may mutate the Raw field.
func (o *ListOptions) AsListOptions() *metav1.ListOptions {
if o == nil {
return &metav1.ListOptions{}
}
if o.Raw == nil {
o.Raw = &metav1.ListOptions{}
}
if o.LabelSelector != nil {
o.Raw.LabelSelector = o.LabelSelector.String()
}
if o.FieldSelector != nil {
o.Raw.FieldSelector = o.FieldSelector.String()
}
if !o.Raw.Watch {
o.Raw.Limit = o.Limit
o.Raw.Continue = o.Continue
}
return o.Raw
}
// ApplyOptions applies the given list options on these options,
// and then returns itself (for convenient chaining).
func (o *ListOptions) ApplyOptions(opts []ListOption) *ListOptions {
for _, opt := range opts {
opt.ApplyToList(o)
}
return o
}
// MatchingLabels filters the list/delete operation on the given set of labels.
type MatchingLabels map[string]string
func (m MatchingLabels) ApplyToList(opts *ListOptions) {
// TODO(directxman12): can we avoid reserializing this over and over?
sel := labels.SelectorFromSet(map[string]string(m))
opts.LabelSelector = sel
}
func (m MatchingLabels) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
m.ApplyToList(&opts.ListOptions)
}
// HasLabels filters the list/delete operation checking if the set of labels exists
// without checking their values.
type HasLabels []string
func (m HasLabels) ApplyToList(opts *ListOptions) {
sel := labels.NewSelector()
for _, label := range m {
r, err := labels.NewRequirement(label, selection.Exists, nil)
if err == nil {
sel = sel.Add(*r)
}
}
opts.LabelSelector = sel
}
func (m HasLabels) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
m.ApplyToList(&opts.ListOptions)
}
// MatchingLabelsSelector filters the list/delete operation on the given label
// selector (or index in the case of cached lists). A struct is used because
// labels.Selector is an interface, which cannot be aliased.
type MatchingLabelsSelector struct {
labels.Selector
}
func (m MatchingLabelsSelector) ApplyToList(opts *ListOptions) {
opts.LabelSelector = m
}
func (m MatchingLabelsSelector) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
m.ApplyToList(&opts.ListOptions)
}
// MatchingField filters the list operation on the given field selector
// (or index in the case of cached lists).
//
// Deprecated: Use MatchingFields
func MatchingField(name, val string) MatchingFields {
return MatchingFields{name: val}
}
// MatchingFields filters the list/delete operation on the given field Set
// (or index in the case of cached lists).
type MatchingFields fields.Set
func (m MatchingFields) ApplyToList(opts *ListOptions) {
// TODO(directxman12): can we avoid re-serializing this?
sel := fields.Set(m).AsSelector()
opts.FieldSelector = sel
}
func (m MatchingFields) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
m.ApplyToList(&opts.ListOptions)
}
// MatchingFieldsSelector filters the list/delete operation on the given field
// selector (or index in the case of cached lists). A struct is used because
// fields.Selector is an interface, which cannot be aliased.
type MatchingFieldsSelector struct {
fields.Selector
}
func (m MatchingFieldsSelector) ApplyToList(opts *ListOptions) {
opts.FieldSelector = m
}
func (m MatchingFieldsSelector) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
m.ApplyToList(&opts.ListOptions)
}
// InNamespace restricts the list/delete operation to the given namespace.
type InNamespace string
func (n InNamespace) ApplyToList(opts *ListOptions) {
opts.Namespace = string(n)
}
func (n InNamespace) ApplyToDeleteAllOf(opts *DeleteAllOfOptions) {
n.ApplyToList(&opts.ListOptions)
}
// Limit specifies the maximum number of results to return from the server.
// Limit does not implement DeleteAllOfOption interface because the server
// does not support setting it for deletecollection operations.
type Limit int64
func (l Limit) ApplyToList(opts *ListOptions) {
opts.Limit = int64(l)
}
// Continue sets a continuation token to retrieve chunks of results when using limit.
// Continue does not implement DeleteAllOfOption interface because the server
// does not support setting it for deletecollection operations.
type Continue string
func (c Continue) ApplyToList(opts *ListOptions) {
opts.Continue = string(c)
}
// }}}
// {{{ Update Options
// UpdateOptions contains options for create requests. It's generally a subset
// of metav1.UpdateOptions.
type UpdateOptions struct {
// When present, indicates that modifications should not be
// persisted. An invalid or unrecognized dryRun directive will
// result in an error response and no further processing of the
// request. Valid values are:
// - All: all dry run stages will be processed
DryRun []string
// FieldManager is the name of the user or component submitting
// this request. It must be set with server-side apply.
FieldManager string
// Raw represents raw UpdateOptions, as passed to the API server.
Raw *metav1.UpdateOptions
}
// AsUpdateOptions returns these options as a metav1.UpdateOptions.
// This may mutate the Raw field.
func (o *UpdateOptions) AsUpdateOptions() *metav1.UpdateOptions {
if o == nil {
return &metav1.UpdateOptions{}
}
if o.Raw == nil {
o.Raw = &metav1.UpdateOptions{}
}
o.Raw.DryRun = o.DryRun
o.Raw.FieldManager = o.FieldManager
return o.Raw
}
// ApplyOptions applies the given update options on these options,
// and then returns itself (for convenient chaining).
func (o *UpdateOptions) ApplyOptions(opts []UpdateOption) *UpdateOptions {
for _, opt := range opts {
opt.ApplyToUpdate(o)
}
return o
}
var _ UpdateOption = &UpdateOptions{}
// ApplyToUpdate implements UpdateOption
func (o *UpdateOptions) ApplyToUpdate(uo *UpdateOptions) {
if o.DryRun != nil {
uo.DryRun = o.DryRun
}
if o.FieldManager != "" {
uo.FieldManager = o.FieldManager
}
if o.Raw != nil {
uo.Raw = o.Raw
}
}
// UpdateDryRunAll sets the "dry run" option to "all".
//
// Deprecated: Use DryRunAll
var UpdateDryRunAll = DryRunAll
// }}}
// {{{ Patch Options
// PatchOptions contains options for patch requests.
type PatchOptions struct {
// When present, indicates that modifications should not be
// persisted. An invalid or unrecognized dryRun directive will
// result in an error response and no further processing of the
// request. Valid values are:
// - All: all dry run stages will be processed
DryRun []string
// Force is going to "force" Apply requests. It means user will
// re-acquire conflicting fields owned by other people. Force
// flag must be unset for non-apply patch requests.
// +optional
Force *bool
// FieldManager is the name of the user or component submitting
// this request. It must be set with server-side apply.
FieldManager string
// Raw represents raw PatchOptions, as passed to the API server.
Raw *metav1.PatchOptions
}
// ApplyOptions applies the given patch options on these options,
// and then returns itself (for convenient chaining).
func (o *PatchOptions) ApplyOptions(opts []PatchOption) *PatchOptions {
for _, opt := range opts {
opt.ApplyToPatch(o)
}
return o
}
// AsPatchOptions returns these options as a metav1.PatchOptions.
// This may mutate the Raw field.
func (o *PatchOptions) AsPatchOptions() *metav1.PatchOptions {
if o == nil {
return &metav1.PatchOptions{}
}
if o.Raw == nil {
o.Raw = &metav1.PatchOptions{}
}
o.Raw.DryRun = o.DryRun
o.Raw.Force = o.Force
o.Raw.FieldManager = o.FieldManager
return o.Raw
}
var _ PatchOption = &PatchOptions{}
// ApplyToPatch implements PatchOptions
func (o *PatchOptions) ApplyToPatch(po *PatchOptions) {
if o.DryRun != nil {
po.DryRun = o.DryRun
}
if o.Force != nil {
po.Force = o.Force
}
if o.FieldManager != "" {
po.FieldManager = o.FieldManager
}
if o.Raw != nil {
po.Raw = o.Raw
}
}
// ForceOwnership indicates that in case of conflicts with server-side apply,
// the client should acquire ownership of the conflicting field. Most
// controllers should use this.
var ForceOwnership = forceOwnership{}
type forceOwnership struct{}
func (forceOwnership) ApplyToPatch(opts *PatchOptions) {
definitelyTrue := true
opts.Force = &definitelyTrue
}
// PatchDryRunAll sets the "dry run" option to "all".
//
// Deprecated: Use DryRunAll
var PatchDryRunAll = DryRunAll
// }}}
// {{{ DeleteAllOf Options
// these are all just delete options and list options
// DeleteAllOfOptions contains options for deletecollection (deleteallof) requests.
// It's just list and delete options smooshed together.
type DeleteAllOfOptions struct {
ListOptions
DeleteOptions
}
// ApplyOptions applies the given deleteallof options on these options,
// and then returns itself (for convenient chaining).
func (o *DeleteAllOfOptions) ApplyOptions(opts []DeleteAllOfOption) *DeleteAllOfOptions {
for _, opt := range opts {
opt.ApplyToDeleteAllOf(o)
}
return o
}
var _ DeleteAllOfOption = &DeleteAllOfOptions{}
// ApplyToDeleteAllOf implements DeleteAllOfOption
func (o *DeleteAllOfOptions) ApplyToDeleteAllOf(do *DeleteAllOfOptions) {
o.ApplyToList(&do.ListOptions)
o.ApplyToDelete(&do.DeleteOptions)
}
// }}}

View file

@ -0,0 +1,123 @@
/*
Copyright 2018 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 client
import (
jsonpatch "github.com/evanphx/json-patch"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/json"
)
var (
// Apply uses server-side apply to patch the given object.
Apply = applyPatch{}
// Merge uses the raw object as a merge patch, without modifications.
// Use MergeFrom if you wish to compute a diff instead.
Merge = mergePatch{}
)
type patch struct {
patchType types.PatchType
data []byte
}
// Type implements Patch.
func (s *patch) Type() types.PatchType {
return s.patchType
}
// Data implements Patch.
func (s *patch) Data(obj runtime.Object) ([]byte, error) {
return s.data, nil
}
// RawPatch constructs a new Patch with the given PatchType and data.
func RawPatch(patchType types.PatchType, data []byte) Patch {
return &patch{patchType, data}
}
// ConstantPatch constructs a new Patch with the given PatchType and data.
//
// Deprecated: use RawPatch instead
func ConstantPatch(patchType types.PatchType, data []byte) Patch {
return RawPatch(patchType, data)
}
type mergeFromPatch struct {
from runtime.Object
}
// Type implements patch.
func (s *mergeFromPatch) Type() types.PatchType {
return types.MergePatchType
}
// Data implements Patch.
func (s *mergeFromPatch) Data(obj runtime.Object) ([]byte, error) {
originalJSON, err := json.Marshal(s.from)
if err != nil {
return nil, err
}
modifiedJSON, err := json.Marshal(obj)
if err != nil {
return nil, err
}
return jsonpatch.CreateMergePatch(originalJSON, modifiedJSON)
}
// MergeFrom creates a Patch that patches using the merge-patch strategy with the given object as base.
func MergeFrom(obj runtime.Object) Patch {
return &mergeFromPatch{obj}
}
// mergePatch uses a raw merge strategy to patch the object.
type mergePatch struct{}
// Type implements Patch.
func (p mergePatch) Type() types.PatchType {
return types.MergePatchType
}
// Data implements Patch.
func (p mergePatch) Data(obj runtime.Object) ([]byte, error) {
// NB(directxman12): we might technically want to be using an actual encoder
// here (in case some more performant encoder is introduced) but this is
// correct and sufficient for our uses (it's what the JSON serializer in
// client-go does, more-or-less).
return json.Marshal(obj)
}
// applyPatch uses server-side apply to patch the object.
type applyPatch struct{}
// Type implements Patch.
func (p applyPatch) Type() types.PatchType {
return types.ApplyPatchType
}
// Data implements Patch.
func (p applyPatch) Data(obj runtime.Object) ([]byte, error) {
// NB(directxman12): we might technically want to be using an actual encoder
// here (in case some more performant encoder is introduced) but this is
// correct and sufficient for our uses (it's what the JSON serializer in
// client-go does, more-or-less).
return json.Marshal(obj)
}

View file

@ -0,0 +1,61 @@
/*
Copyright 2018 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 client
import (
"context"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)
// DelegatingClient forms a Client by composing separate reader, writer and
// statusclient interfaces. This way, you can have an Client that reads from a
// cache and writes to the API server.
type DelegatingClient struct {
Reader
Writer
StatusClient
}
// DelegatingReader forms a Reader that will cause Get and List requests for
// unstructured types to use the ClientReader while requests for any other type
// of object with use the CacheReader. This avoids accidentally caching the
// entire cluster in the common case of loading arbitrary unstructured objects
// (e.g. from OwnerReferences).
type DelegatingReader struct {
CacheReader Reader
ClientReader Reader
}
// Get retrieves an obj for a given object key from the Kubernetes Cluster.
func (d *DelegatingReader) Get(ctx context.Context, key ObjectKey, obj runtime.Object) error {
_, isUnstructured := obj.(*unstructured.Unstructured)
if isUnstructured {
return d.ClientReader.Get(ctx, key, obj)
}
return d.CacheReader.Get(ctx, key, obj)
}
// List retrieves list of objects for a given namespace and list options.
func (d *DelegatingReader) List(ctx context.Context, list runtime.Object, opts ...ListOption) error {
_, isUnstructured := list.(*unstructured.UnstructuredList)
if isUnstructured {
return d.ClientReader.List(ctx, list, opts...)
}
return d.CacheReader.List(ctx, list, opts...)
}

View file

@ -0,0 +1,201 @@
/*
Copyright 2018 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 client
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
)
// client is a client.Client that reads and writes directly from/to an API server. It lazily initializes
// new clients at the time they are used, and caches the client.
type typedClient struct {
cache *clientCache
paramCodec runtime.ParameterCodec
}
// Create implements client.Client
func (c *typedClient) Create(ctx context.Context, obj runtime.Object, opts ...CreateOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
createOpts := &CreateOptions{}
createOpts.ApplyOptions(opts)
return o.Post().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Body(obj).
VersionedParams(createOpts.AsCreateOptions(), c.paramCodec).
Do(ctx).
Into(obj)
}
// Update implements client.Client
func (c *typedClient) Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
updateOpts := &UpdateOptions{}
updateOpts.ApplyOptions(opts)
return o.Put().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
Body(obj).
VersionedParams(updateOpts.AsUpdateOptions(), c.paramCodec).
Do(ctx).
Into(obj)
}
// Delete implements client.Client
func (c *typedClient) Delete(ctx context.Context, obj runtime.Object, opts ...DeleteOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
deleteOpts := DeleteOptions{}
deleteOpts.ApplyOptions(opts)
return o.Delete().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
Body(deleteOpts.AsDeleteOptions()).
Do(ctx).
Error()
}
// DeleteAllOf implements client.Client
func (c *typedClient) DeleteAllOf(ctx context.Context, obj runtime.Object, opts ...DeleteAllOfOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
deleteAllOfOpts := DeleteAllOfOptions{}
deleteAllOfOpts.ApplyOptions(opts)
return o.Delete().
NamespaceIfScoped(deleteAllOfOpts.ListOptions.Namespace, o.isNamespaced()).
Resource(o.resource()).
VersionedParams(deleteAllOfOpts.AsListOptions(), c.paramCodec).
Body(deleteAllOfOpts.AsDeleteOptions()).
Do(ctx).
Error()
}
// Patch implements client.Client
func (c *typedClient) Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
data, err := patch.Data(obj)
if err != nil {
return err
}
patchOpts := &PatchOptions{}
return o.Patch(patch.Type()).
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
VersionedParams(patchOpts.ApplyOptions(opts).AsPatchOptions(), c.paramCodec).
Body(data).
Do(ctx).
Into(obj)
}
// Get implements client.Client
func (c *typedClient) Get(ctx context.Context, key ObjectKey, obj runtime.Object) error {
r, err := c.cache.getResource(obj)
if err != nil {
return err
}
return r.Get().
NamespaceIfScoped(key.Namespace, r.isNamespaced()).
Resource(r.resource()).
Name(key.Name).Do(ctx).Into(obj)
}
// List implements client.Client
func (c *typedClient) List(ctx context.Context, obj runtime.Object, opts ...ListOption) error {
r, err := c.cache.getResource(obj)
if err != nil {
return err
}
listOpts := ListOptions{}
listOpts.ApplyOptions(opts)
return r.Get().
NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
Resource(r.resource()).
VersionedParams(listOpts.AsListOptions(), c.paramCodec).
Do(ctx).
Into(obj)
}
// UpdateStatus used by StatusWriter to write status.
func (c *typedClient) UpdateStatus(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
// TODO(droot): examine the returned error and check if it error needs to be
// wrapped to improve the UX ?
// It will be nice to receive an error saying the object doesn't implement
// status subresource and check CRD definition
return o.Put().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
SubResource("status").
Body(obj).
VersionedParams((&UpdateOptions{}).ApplyOptions(opts).AsUpdateOptions(), c.paramCodec).
Do(ctx).
Into(obj)
}
// PatchStatus used by StatusWriter to write status.
func (c *typedClient) PatchStatus(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
o, err := c.cache.getObjMeta(obj)
if err != nil {
return err
}
data, err := patch.Data(obj)
if err != nil {
return err
}
patchOpts := &PatchOptions{}
return o.Patch(patch.Type()).
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
SubResource("status").
Body(data).
VersionedParams(patchOpts.ApplyOptions(opts).AsPatchOptions(), c.paramCodec).
Do(ctx).
Into(obj)
}

View file

@ -0,0 +1,273 @@
/*
Copyright 2018 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 client
import (
"context"
"fmt"
"strings"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)
// client is a client.Client that reads and writes directly from/to an API server. It lazily initializes
// new clients at the time they are used, and caches the client.
type unstructuredClient struct {
cache *clientCache
paramCodec runtime.ParameterCodec
}
// Create implements client.Client
func (uc *unstructuredClient) Create(ctx context.Context, obj runtime.Object, opts ...CreateOption) error {
u, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
gvk := u.GroupVersionKind()
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
createOpts := &CreateOptions{}
createOpts.ApplyOptions(opts)
result := o.Post().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Body(obj).
VersionedParams(createOpts.AsCreateOptions(), uc.paramCodec).
Do(ctx).
Into(obj)
u.SetGroupVersionKind(gvk)
return result
}
// Update implements client.Client
func (uc *unstructuredClient) Update(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
u, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
gvk := u.GroupVersionKind()
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
updateOpts := UpdateOptions{}
updateOpts.ApplyOptions(opts)
result := o.Put().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
Body(obj).
VersionedParams(updateOpts.AsUpdateOptions(), uc.paramCodec).
Do(ctx).
Into(obj)
u.SetGroupVersionKind(gvk)
return result
}
// Delete implements client.Client
func (uc *unstructuredClient) Delete(ctx context.Context, obj runtime.Object, opts ...DeleteOption) error {
_, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
deleteOpts := DeleteOptions{}
deleteOpts.ApplyOptions(opts)
return o.Delete().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
Body(deleteOpts.AsDeleteOptions()).
Do(ctx).
Error()
}
// DeleteAllOf implements client.Client
func (uc *unstructuredClient) DeleteAllOf(ctx context.Context, obj runtime.Object, opts ...DeleteAllOfOption) error {
_, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
deleteAllOfOpts := DeleteAllOfOptions{}
deleteAllOfOpts.ApplyOptions(opts)
return o.Delete().
NamespaceIfScoped(deleteAllOfOpts.ListOptions.Namespace, o.isNamespaced()).
Resource(o.resource()).
VersionedParams(deleteAllOfOpts.AsListOptions(), uc.paramCodec).
Body(deleteAllOfOpts.AsDeleteOptions()).
Do(ctx).
Error()
}
// Patch implements client.Client
func (uc *unstructuredClient) Patch(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
_, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
data, err := patch.Data(obj)
if err != nil {
return err
}
patchOpts := &PatchOptions{}
return o.Patch(patch.Type()).
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
VersionedParams(patchOpts.ApplyOptions(opts).AsPatchOptions(), uc.paramCodec).
Body(data).
Do(ctx).
Into(obj)
}
// Get implements client.Client
func (uc *unstructuredClient) Get(ctx context.Context, key ObjectKey, obj runtime.Object) error {
u, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
gvk := u.GroupVersionKind()
r, err := uc.cache.getResource(obj)
if err != nil {
return err
}
result := r.Get().
NamespaceIfScoped(key.Namespace, r.isNamespaced()).
Resource(r.resource()).
Name(key.Name).
Do(ctx).
Into(obj)
u.SetGroupVersionKind(gvk)
return result
}
// List implements client.Client
func (uc *unstructuredClient) List(ctx context.Context, obj runtime.Object, opts ...ListOption) error {
u, ok := obj.(*unstructured.UnstructuredList)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
gvk := u.GroupVersionKind()
if strings.HasSuffix(gvk.Kind, "List") {
gvk.Kind = gvk.Kind[:len(gvk.Kind)-4]
}
listOpts := ListOptions{}
listOpts.ApplyOptions(opts)
r, err := uc.cache.getResource(obj)
if err != nil {
return err
}
return r.Get().
NamespaceIfScoped(listOpts.Namespace, r.isNamespaced()).
Resource(r.resource()).
VersionedParams(listOpts.AsListOptions(), uc.paramCodec).
Do(ctx).
Into(obj)
}
func (uc *unstructuredClient) UpdateStatus(ctx context.Context, obj runtime.Object, opts ...UpdateOption) error {
_, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
return o.Put().
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
SubResource("status").
Body(obj).
VersionedParams((&UpdateOptions{}).ApplyOptions(opts).AsUpdateOptions(), uc.paramCodec).
Do(ctx).
Into(obj)
}
func (uc *unstructuredClient) PatchStatus(ctx context.Context, obj runtime.Object, patch Patch, opts ...PatchOption) error {
u, ok := obj.(*unstructured.Unstructured)
if !ok {
return fmt.Errorf("unstructured client did not understand object: %T", obj)
}
gvk := u.GroupVersionKind()
o, err := uc.cache.getObjMeta(obj)
if err != nil {
return err
}
data, err := patch.Data(obj)
if err != nil {
return err
}
patchOpts := &PatchOptions{}
result := o.Patch(patch.Type()).
NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
Resource(o.resource()).
Name(o.GetName()).
SubResource("status").
Body(data).
VersionedParams(patchOpts.ApplyOptions(opts).AsPatchOptions(), uc.paramCodec).
Do(ctx).
Into(u)
u.SetGroupVersionKind(gvk)
return result
}

View file

@ -19,29 +19,33 @@ package envtest
import (
"bufio"
"bytes"
"context"
"io"
"io/ioutil"
"os"
"path/filepath"
"time"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
k8syaml "k8s.io/apimachinery/pkg/util/yaml"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/yaml"
)
// CRDInstallOptions are the options for installing CRDs
type CRDInstallOptions struct {
// Paths is a list of paths to the directories containing CRDs
// Paths is a list of paths to the directories or files containing CRDs
Paths []string
// CRDs is a list of CRDs to install
CRDs []*apiextensionsv1beta1.CustomResourceDefinition
CRDs []runtime.Object
// ErrorIfPathMissing will cause an error if a Path does not exist
ErrorIfPathMissing bool
@ -51,13 +55,18 @@ type CRDInstallOptions struct {
// PollInterval is the interval to check
PollInterval time.Duration
// CleanUpAfterUse will cause the CRDs listed for installation to be
// uninstalled when terminating the test environment.
// Defaults to false.
CleanUpAfterUse bool
}
const defaultPollInterval = 100 * time.Millisecond
const defaultMaxWait = 10 * time.Second
// InstallCRDs installs a collection of CRDs into a cluster by reading the crd yaml files from a directory
func InstallCRDs(config *rest.Config, options CRDInstallOptions) ([]*apiextensionsv1beta1.CustomResourceDefinition, error) {
func InstallCRDs(config *rest.Config, options CRDInstallOptions) ([]runtime.Object, error) {
defaultCRDOptions(&options)
// Read the CRD yamls into options.CRDs
@ -81,16 +90,12 @@ func InstallCRDs(config *rest.Config, options CRDInstallOptions) ([]*apiextensio
// readCRDFiles reads the directories of CRDs in options.Paths and adds the CRD structs to options.CRDs
func readCRDFiles(options *CRDInstallOptions) error {
if len(options.Paths) > 0 {
for _, path := range options.Paths {
if _, err := os.Stat(path); !options.ErrorIfPathMissing && os.IsNotExist(err) {
continue
}
new, err := readCRDs(path)
if err != nil {
return err
}
options.CRDs = append(options.CRDs, new...)
crdList, err := renderCRDs(options)
if err != nil {
return err
}
options.CRDs = append(options.CRDs, crdList...)
}
return nil
}
@ -106,19 +111,49 @@ func defaultCRDOptions(o *CRDInstallOptions) {
}
// WaitForCRDs waits for the CRDs to appear in discovery
func WaitForCRDs(config *rest.Config, crds []*apiextensionsv1beta1.CustomResourceDefinition, options CRDInstallOptions) error {
func WaitForCRDs(config *rest.Config, crds []runtime.Object, options CRDInstallOptions) error {
// Add each CRD to a map of GroupVersion to Resource
waitingFor := map[schema.GroupVersion]*sets.String{}
for _, crd := range crds {
for _, crd := range runtimeCRDListToUnstructured(crds) {
gvs := []schema.GroupVersion{}
if crd.Spec.Version != "" {
gvs = append(gvs, schema.GroupVersion{Group: crd.Spec.Group, Version: crd.Spec.Version})
crdGroup, _, err := unstructured.NestedString(crd.Object, "spec", "group")
if err != nil {
return err
}
for _, ver := range crd.Spec.Versions {
if ver.Served {
gvs = append(gvs, schema.GroupVersion{Group: crd.Spec.Group, Version: ver.Name})
crdPlural, _, err := unstructured.NestedString(crd.Object, "spec", "names", "plural")
if err != nil {
return err
}
crdVersion, _, err := unstructured.NestedString(crd.Object, "spec", "version")
if err != nil {
return err
}
if crdVersion != "" {
gvs = append(gvs, schema.GroupVersion{Group: crdGroup, Version: crdVersion})
}
versions, _, err := unstructured.NestedSlice(crd.Object, "spec", "versions")
if err != nil {
return err
}
for _, version := range versions {
versionMap, ok := version.(map[string]interface{})
if !ok {
continue
}
served, _, err := unstructured.NestedBool(versionMap, "served")
if err != nil {
return err
}
if served {
versionName, _, err := unstructured.NestedString(versionMap, "name")
if err != nil {
return err
}
gvs = append(gvs, schema.GroupVersion{Group: crdGroup, Version: versionName})
}
}
for _, gv := range gvs {
log.V(1).Info("adding API in waitlist", "GV", gv)
if _, found := waitingFor[gv]; !found {
@ -126,7 +161,7 @@ func WaitForCRDs(config *rest.Config, crds []*apiextensionsv1beta1.CustomResourc
waitingFor[gv] = &sets.String{}
}
// Add the Resource
waitingFor[gv].Insert(crd.Spec.Names.Plural)
waitingFor[gv].Insert(crdPlural)
}
}
@ -180,37 +215,115 @@ func (p *poller) poll() (done bool, err error) {
return allFound, nil
}
// UninstallCRDs uninstalls a collection of CRDs by reading the crd yaml files from a directory
func UninstallCRDs(config *rest.Config, options CRDInstallOptions) error {
// Read the CRD yamls into options.CRDs
if err := readCRDFiles(&options); err != nil {
return err
}
// Delete the CRDs from the apiserver
cs, err := client.New(config, client.Options{})
if err != nil {
return err
}
// Uninstall each CRD
for _, crd := range runtimeCRDListToUnstructured(options.CRDs) {
log.V(1).Info("uninstalling CRD", "crd", crd.GetName())
if err := cs.Delete(context.TODO(), crd); err != nil {
// If CRD is not found, we can consider success
if !apierrors.IsNotFound(err) {
return err
}
}
}
return nil
}
// CreateCRDs creates the CRDs
func CreateCRDs(config *rest.Config, crds []*apiextensionsv1beta1.CustomResourceDefinition) error {
cs, err := clientset.NewForConfig(config)
func CreateCRDs(config *rest.Config, crds []runtime.Object) error {
cs, err := client.New(config, client.Options{})
if err != nil {
return err
}
// Create each CRD
for _, crd := range crds {
log.V(1).Info("installing CRD", "crd", crd.Name)
if _, err := cs.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd); err != nil {
for _, crd := range runtimeCRDListToUnstructured(crds) {
log.V(1).Info("installing CRD", "crd", crd.GetName())
existingCrd := crd.DeepCopy()
err := cs.Get(context.TODO(), client.ObjectKey{Name: crd.GetName()}, existingCrd)
switch {
case apierrors.IsNotFound(err):
if err := cs.Create(context.TODO(), crd); err != nil {
return err
}
case err != nil:
return err
default:
log.V(1).Info("CRD already exists, updating", "crd", crd.GetName())
crd.SetResourceVersion(existingCrd.GetResourceVersion())
if err := cs.Update(context.TODO(), crd); err != nil {
return err
}
}
}
return nil
}
// readCRDs reads the CRDs from files and Unmarshals them into structs
func readCRDs(path string) ([]*apiextensionsv1beta1.CustomResourceDefinition, error) {
// Get the CRD files
var files []os.FileInfo
var err error
log.V(1).Info("reading CRDs from path", "path", path)
if files, err = ioutil.ReadDir(path); err != nil {
return nil, err
// renderCRDs iterate through options.Paths and extract all CRD files.
func renderCRDs(options *CRDInstallOptions) ([]runtime.Object, error) {
var (
err error
info os.FileInfo
crds []*unstructured.Unstructured
files []os.FileInfo
)
for _, path := range options.Paths {
var filePath = path
// Return the error if ErrorIfPathMissing exists
if info, err = os.Stat(path); os.IsNotExist(err) {
if options.ErrorIfPathMissing {
return nil, err
}
continue
}
if !info.IsDir() {
filePath, files = filepath.Dir(path), append(files, info)
} else {
if files, err = ioutil.ReadDir(path); err != nil {
return nil, err
}
}
log.V(1).Info("reading CRDs from path", "path", path)
crdList, err := readCRDs(filePath, files)
if err != nil {
return nil, err
}
// If CRD already in the list, skip it.
if existsUnstructured(crds, crdList) {
continue
}
crds = append(crds, crdList...)
}
return unstructuredCRDListToRuntime(crds), nil
}
// readCRDs reads the CRDs from files and Unmarshals them into structs
func readCRDs(basePath string, files []os.FileInfo) ([]*unstructured.Unstructured, error) {
var crds []*unstructured.Unstructured
// White list the file extensions that may contain CRDs
crdExts := sets.NewString(".json", ".yaml", ".yml")
var crds []*apiextensionsv1beta1.CustomResourceDefinition
for _, file := range files {
// Only parse whitelisted file types
if !crdExts.Has(filepath.Ext(file.Name())) {
@ -218,19 +331,28 @@ func readCRDs(path string) ([]*apiextensionsv1beta1.CustomResourceDefinition, er
}
// Unmarshal CRDs from file into structs
docs, err := readDocuments(filepath.Join(path, file.Name()))
docs, err := readDocuments(filepath.Join(basePath, file.Name()))
if err != nil {
return nil, err
}
for _, doc := range docs {
crd := &apiextensionsv1beta1.CustomResourceDefinition{}
crd := &unstructured.Unstructured{}
if err = yaml.Unmarshal(doc, crd); err != nil {
return nil, err
}
// Check that it is actually a CRD
if crd.Spec.Names.Kind == "" || crd.Spec.Group == "" {
crdKind, _, err := unstructured.NestedString(crd.Object, "spec", "names", "kind")
if err != nil {
return nil, err
}
crdGroup, _, err := unstructured.NestedString(crd.Object, "spec", "group")
if err != nil {
return nil, err
}
if crd.GetKind() != "CustomResourceDefinition" || crdKind == "" || crdGroup == "" {
continue
}
crds = append(crds, crd)

View file

@ -1,11 +0,0 @@
package envtest
import (
"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
)
// NewlineReporter is Reporter that Prints a newline after the default Reporter output so that the results
// are correctly parsed by test automation.
// See issue https://github.com/jstemmer/go-junit-report/issues/31
// It's re-exported here to avoid compatibility breakage/mass rewrites.
type NewlineReporter = printer.NewlineReporter

View file

@ -1,6 +1,23 @@
package envtest
import apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
import (
"reflect"
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)
var (
crdScheme = runtime.NewScheme()
)
// init is required to correctly initialize the crdScheme package variable.
func init() {
_ = apiextensionsv1.AddToScheme(crdScheme)
_ = apiextensionsv1beta1.AddToScheme(crdScheme)
}
// mergePaths merges two string slices containing paths.
// This function makes no guarantees about order of the merged slice.
@ -23,19 +40,52 @@ func mergePaths(s1, s2 []string) []string {
// mergeCRDs merges two CRD slices using their names.
// This function makes no guarantees about order of the merged slice.
func mergeCRDs(s1, s2 []*apiextensionsv1beta1.CustomResourceDefinition) []*apiextensionsv1beta1.CustomResourceDefinition {
m := make(map[string]*apiextensionsv1beta1.CustomResourceDefinition)
for _, crd := range s1 {
m[crd.Name] = crd
func mergeCRDs(s1, s2 []runtime.Object) []runtime.Object {
m := make(map[string]*unstructured.Unstructured)
for _, obj := range runtimeCRDListToUnstructured(s1) {
m[obj.GetName()] = obj
}
for _, crd := range s2 {
m[crd.Name] = crd
for _, obj := range runtimeCRDListToUnstructured(s2) {
m[obj.GetName()] = obj
}
merged := make([]*apiextensionsv1beta1.CustomResourceDefinition, len(m))
merged := make([]runtime.Object, len(m))
i := 0
for _, crd := range m {
merged[i] = crd
for _, obj := range m {
merged[i] = obj
i++
}
return merged
}
// existsUnstructured verify if a any item is common between two lists.
func existsUnstructured(s1, s2 []*unstructured.Unstructured) bool {
for _, s1obj := range s1 {
for _, s2obj := range s2 {
if reflect.DeepEqual(s1obj, s2obj) {
return true
}
}
}
return false
}
func runtimeCRDListToUnstructured(l []runtime.Object) []*unstructured.Unstructured {
res := []*unstructured.Unstructured{}
for _, obj := range l {
u := &unstructured.Unstructured{}
if err := crdScheme.Convert(obj, u, nil); err != nil {
log.Error(err, "error converting to unstructured object", "object-kind", obj.GetObjectKind())
continue
}
res = append(res, u)
}
return res
}
func unstructuredCRDListToRuntime(l []*unstructured.Unstructured) []runtime.Object {
res := []runtime.Object{}
for _, obj := range l {
res = append(res, obj.DeepCopy())
}
return res
}

View file

@ -1,53 +0,0 @@
/*
Copyright 2016 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 printer contains setup for a friendlier Ginkgo printer that's easier
// to parse by test automation.
package printer
import (
"fmt"
"github.com/onsi/ginkgo"
"github.com/onsi/ginkgo/config"
"github.com/onsi/ginkgo/types"
)
var _ ginkgo.Reporter = NewlineReporter{}
// NewlineReporter is Reporter that Prints a newline after the default Reporter output so that the results
// are correctly parsed by test automation.
// See issue https://github.com/jstemmer/go-junit-report/issues/31
type NewlineReporter struct{}
// SpecSuiteWillBegin implements ginkgo.Reporter
func (NewlineReporter) SpecSuiteWillBegin(config config.GinkgoConfigType, summary *types.SuiteSummary) {
}
// BeforeSuiteDidRun implements ginkgo.Reporter
func (NewlineReporter) BeforeSuiteDidRun(setupSummary *types.SetupSummary) {}
// AfterSuiteDidRun implements ginkgo.Reporter
func (NewlineReporter) AfterSuiteDidRun(setupSummary *types.SetupSummary) {}
// SpecWillRun implements ginkgo.Reporter
func (NewlineReporter) SpecWillRun(specSummary *types.SpecSummary) {}
// SpecDidComplete implements ginkgo.Reporter
func (NewlineReporter) SpecDidComplete(specSummary *types.SpecSummary) {}
// SpecSuiteDidEnd Prints a newline between "35 Passed | 0 Failed | 0 Pending | 0 Skipped" and "--- PASS:"
func (NewlineReporter) SpecSuiteDidEnd(summary *types.SuiteSummary) { fmt.Printf("\n") }

View file

@ -23,10 +23,10 @@ import (
"strings"
"time"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/client/config"
"sigs.k8s.io/testing_frameworks/integration"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration"
logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
)
@ -72,20 +72,6 @@ func defaultAssetPath(binary string) string {
}
// DefaultKubeAPIServerFlags are default flags necessary to bring up apiserver.
var DefaultKubeAPIServerFlags = []string{
// Allow tests to run offline, by preventing API server from attempting to
// use default route to determine its --advertise-address
"--advertise-address=127.0.0.1",
"--etcd-servers={{ if .EtcdURL }}{{ .EtcdURL.String }}{{ end }}",
"--cert-dir={{ .CertDir }}",
"--insecure-port={{ if .URL }}{{ .URL.Port }}{{ end }}",
"--insecure-bind-address={{ if .URL }}{{ .URL.Hostname }}{{ end }}",
"--secure-port={{ if .SecurePort }}{{ .SecurePort }}{{ end }}",
"--admission-control=AlwaysAdmit",
"--service-cluster-ip-range=10.0.0.0/24",
}
// Environment creates a Kubernetes test environment that will start / stop the Kubernetes control plane and
// install extension APIs
type Environment struct {
@ -100,10 +86,18 @@ type Environment struct {
// CRDInstallOptions are the options for installing CRDs.
CRDInstallOptions CRDInstallOptions
// CRDInstallOptions are the options for installing webhooks.
WebhookInstallOptions WebhookInstallOptions
// ErrorIfCRDPathMissing provides an interface for the underlying
// CRDInstallOptions.ErrorIfPathMissing. It prevents silent failures
// for missing CRD paths.
ErrorIfCRDPathMissing bool
// CRDs is a list of CRDs to install.
// If both this field and CRDs field in CRDInstallOptions are specified, the
// values are merged.
CRDs []*apiextensionsv1beta1.CustomResourceDefinition
CRDs []runtime.Object
// CRDDirectoryPaths is a list of paths containing CRD yaml or json configs.
// If both this field and Paths field in CRDInstallOptions are specified, the
@ -135,19 +129,30 @@ type Environment struct {
}
// Stop stops a running server.
// If USE_EXISTING_CLUSTER is set to true, this method is a no-op.
// Previously installed CRDs, as listed in CRDInstallOptions.CRDs, will be uninstalled
// if CRDInstallOptions.CleanUpAfterUse are set to true.
func (te *Environment) Stop() error {
if te.CRDInstallOptions.CleanUpAfterUse {
if err := UninstallCRDs(te.Config, te.CRDInstallOptions); err != nil {
return err
}
}
if te.useExistingCluster() {
return nil
}
err := te.WebhookInstallOptions.Cleanup()
if err != nil {
return err
}
return te.ControlPlane.Stop()
}
// getAPIServerFlags returns flags to be used with the Kubernetes API server.
// it returns empty slice for api server defined defaults to be applied if no args specified
func (te Environment) getAPIServerFlags() []string {
// Set default API server flags if not set.
if len(te.KubeAPIServerFlags) == 0 {
return DefaultKubeAPIServerFlags
return []string{}
}
// Check KubeAPIServerFlags contains service-cluster-ip-range, if not, set default value to service-cluster-ip-range
containServiceClusterIPRange := false
@ -216,7 +221,7 @@ func (te *Environment) Start() (*rest.Config, error) {
}
if err := te.defaultTimeouts(); err != nil {
return nil, fmt.Errorf("failed to default controlplane timeouts: %v", err)
return nil, fmt.Errorf("failed to default controlplane timeouts: %w", err)
}
te.ControlPlane.Etcd.StartTimeout = te.ControlPlaneStartTimeout
te.ControlPlane.Etcd.StopTimeout = te.ControlPlaneStopTimeout
@ -240,7 +245,16 @@ func (te *Environment) Start() (*rest.Config, error) {
log.V(1).Info("installing CRDs")
te.CRDInstallOptions.CRDs = mergeCRDs(te.CRDInstallOptions.CRDs, te.CRDs)
te.CRDInstallOptions.Paths = mergePaths(te.CRDInstallOptions.Paths, te.CRDDirectoryPaths)
_, err := InstallCRDs(te.Config, te.CRDInstallOptions)
te.CRDInstallOptions.ErrorIfPathMissing = te.ErrorIfCRDPathMissing
crds, err := InstallCRDs(te.Config, te.CRDInstallOptions)
if err != nil {
return te.Config, err
}
te.CRDs = crds
log.V(1).Info("installing webhooks")
err = te.WebhookInstallOptions.Install(te.Config)
return te.Config, err
}
@ -256,7 +270,7 @@ func (te *Environment) startControlPlane() error {
log.Error(err, "unable to start the controlplane", "tries", numTries)
}
if numTries == maxRetries {
return fmt.Errorf("failed to start the controlplane. retried %d times: %v", numTries, err)
return fmt.Errorf("failed to start the controlplane. retried %d times: %w", numTries, err)
}
return nil
}
@ -293,3 +307,7 @@ func (te *Environment) useExistingCluster() bool {
}
return *te.UseExistingCluster
}
// DefaultKubeAPIServerFlags exposes the default args for the APIServer so that
// you can use those to append your own additional arguments.
var DefaultKubeAPIServerFlags = integration.APIServerDefaultArgs

View file

@ -0,0 +1,423 @@
/*
Copyright 2019 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 envtest
import (
"context"
"encoding/base64"
"fmt"
"io/ioutil"
"net"
"os"
"path/filepath"
"time"
"k8s.io/apimachinery/pkg/api/errors"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/addr"
"sigs.k8s.io/yaml"
)
// WebhookInstallOptions are the options for installing mutating or validating webhooks
type WebhookInstallOptions struct {
// Paths is a list of paths to the directories containing the mutating or validating webhooks yaml or json configs.
DirectoryPaths []string
// MutatingWebhooks is a list of MutatingWebhookConfigurations to install
MutatingWebhooks []runtime.Object
// ValidatingWebhooks is a list of ValidatingWebhookConfigurations to install
ValidatingWebhooks []runtime.Object
// IgnoreErrorIfPathMissing will ignore an error if a DirectoryPath does not exist when set to true
IgnoreErrorIfPathMissing bool
// LocalServingHost is the host for serving webhooks on.
// it will be automatically populated
LocalServingHost string
// LocalServingPort is the allocated port for serving webhooks on.
// it will be automatically populated by a random available local port
LocalServingPort int
// LocalServingCertDir is the allocated directory for serving certificates.
// it will be automatically populated by the local temp dir
LocalServingCertDir string
// MaxTime is the max time to wait
MaxTime time.Duration
// PollInterval is the interval to check
PollInterval time.Duration
}
// ModifyWebhookDefinitions modifies webhook definitions by:
// - applying CABundle based on the provided tinyca
// - if webhook client config uses service spec, it's removed and replaced with direct url
func (o *WebhookInstallOptions) ModifyWebhookDefinitions(caData []byte) error {
hostPort, err := o.generateHostPort()
if err != nil {
return err
}
for i, unstructuredHook := range runtimeListToUnstructured(o.MutatingWebhooks) {
webhooks, found, err := unstructured.NestedSlice(unstructuredHook.Object, "webhooks")
if !found || err != nil {
return fmt.Errorf("unexpected object, %v", err)
}
for j := range webhooks {
webhook, err := modifyWebhook(webhooks[j].(map[string]interface{}), caData, hostPort)
if err != nil {
return err
}
webhooks[j] = webhook
unstructuredHook.Object["webhooks"] = webhooks
o.MutatingWebhooks[i] = unstructuredHook
}
}
for i, unstructuredHook := range runtimeListToUnstructured(o.ValidatingWebhooks) {
webhooks, found, err := unstructured.NestedSlice(unstructuredHook.Object, "webhooks")
if !found || err != nil {
return fmt.Errorf("unexpected object, %v", err)
}
for j := range webhooks {
webhook, err := modifyWebhook(webhooks[j].(map[string]interface{}), caData, hostPort)
if err != nil {
return err
}
webhooks[j] = webhook
unstructuredHook.Object["webhooks"] = webhooks
o.ValidatingWebhooks[i] = unstructuredHook
}
}
return nil
}
func modifyWebhook(webhook map[string]interface{}, caData []byte, hostPort string) (map[string]interface{}, error) {
clientConfig, found, err := unstructured.NestedMap(webhook, "clientConfig")
if !found || err != nil {
return nil, fmt.Errorf("cannot find clientconfig: %v", err)
}
clientConfig["caBundle"] = base64.StdEncoding.EncodeToString(caData)
servicePath, found, err := unstructured.NestedString(clientConfig, "service", "path")
if found && err == nil {
// we cannot use service in integration tests since we're running controller outside cluster
// the intent here is that we swap out service for raw address because we don't have an actually standard kube service network.
// We want to users to be able to use your standard config though
url := fmt.Sprintf("https://%s/%s", hostPort, servicePath)
clientConfig["url"] = url
clientConfig["service"] = nil
}
webhook["clientConfig"] = clientConfig
return webhook, nil
}
func (o *WebhookInstallOptions) generateHostPort() (string, error) {
port, host, err := addr.Suggest()
if err != nil {
return "", fmt.Errorf("unable to grab random port for serving webhooks on: %v", err)
}
o.LocalServingPort = port
o.LocalServingHost = host
return net.JoinHostPort(host, fmt.Sprintf("%d", port)), nil
}
// Install installs specified webhooks to the API server
func (o *WebhookInstallOptions) Install(config *rest.Config) error {
hookCA, err := o.setupCA()
if err != nil {
return err
}
if err := parseWebhookDirs(o); err != nil {
return err
}
err = o.ModifyWebhookDefinitions(hookCA)
if err != nil {
return err
}
if err := createWebhooks(config, o.MutatingWebhooks, o.ValidatingWebhooks); err != nil {
return err
}
if err := WaitForWebhooks(config, o.MutatingWebhooks, o.ValidatingWebhooks, *o); err != nil {
return err
}
return nil
}
// Cleanup cleans up cert directories
func (o *WebhookInstallOptions) Cleanup() error {
if o.LocalServingCertDir != "" {
return os.RemoveAll(o.LocalServingCertDir)
}
return nil
}
// WaitForWebhooks waits for the Webhooks to be available through API server
func WaitForWebhooks(config *rest.Config,
mutatingWebhooks []runtime.Object,
validatingWebhooks []runtime.Object,
options WebhookInstallOptions) error {
waitingFor := map[schema.GroupVersionKind]*sets.String{}
for _, hook := range runtimeListToUnstructured(append(validatingWebhooks, mutatingWebhooks...)) {
if _, ok := waitingFor[hook.GroupVersionKind()]; !ok {
waitingFor[hook.GroupVersionKind()] = &sets.String{}
}
waitingFor[hook.GroupVersionKind()].Insert(hook.GetName())
}
// Poll until all resources are found in discovery
p := &webhookPoller{config: config, waitingFor: waitingFor}
return wait.PollImmediate(options.PollInterval, options.MaxTime, p.poll)
}
// poller checks if all the resources have been found in discovery, and returns false if not
type webhookPoller struct {
// config is used to get discovery
config *rest.Config
// waitingFor is the map of resources keyed by group version that have not yet been found in discovery
waitingFor map[schema.GroupVersionKind]*sets.String
}
// poll checks if all the resources have been found in discovery, and returns false if not
func (p *webhookPoller) poll() (done bool, err error) {
// Create a new clientset to avoid any client caching of discovery
c, err := client.New(p.config, client.Options{})
if err != nil {
return false, err
}
allFound := true
for gvk, names := range p.waitingFor {
if names.Len() == 0 {
delete(p.waitingFor, gvk)
continue
}
for _, name := range names.List() {
var obj = &unstructured.Unstructured{}
obj.SetGroupVersionKind(gvk)
err := c.Get(context.Background(), client.ObjectKey{
Namespace: "",
Name: name,
}, obj)
if err == nil {
names.Delete(name)
}
if errors.IsNotFound(err) {
allFound = false
}
if err != nil {
return false, err
}
}
}
return allFound, nil
}
// setupCA creates CA for testing and writes them to disk
func (o *WebhookInstallOptions) setupCA() ([]byte, error) {
hookCA, err := integration.NewTinyCA()
if err != nil {
return nil, fmt.Errorf("unable to set up webhook CA: %v", err)
}
hookCert, err := hookCA.NewServingCert()
if err != nil {
return nil, fmt.Errorf("unable to set up webhook serving certs: %v", err)
}
localServingCertsDir, err := ioutil.TempDir("", "envtest-serving-certs-")
o.LocalServingCertDir = localServingCertsDir
if err != nil {
return nil, fmt.Errorf("unable to create directory for webhook serving certs: %v", err)
}
certData, keyData, err := hookCert.AsBytes()
if err != nil {
return nil, fmt.Errorf("unable to marshal webhook serving certs: %v", err)
}
if err := ioutil.WriteFile(filepath.Join(localServingCertsDir, "tls.crt"), certData, 0640); err != nil {
return nil, fmt.Errorf("unable to write webhook serving cert to disk: %v", err)
}
if err := ioutil.WriteFile(filepath.Join(localServingCertsDir, "tls.key"), keyData, 0640); err != nil {
return nil, fmt.Errorf("unable to write webhook serving key to disk: %v", err)
}
return certData, nil
}
func createWebhooks(config *rest.Config, mutHooks []runtime.Object, valHooks []runtime.Object) error {
cs, err := client.New(config, client.Options{})
if err != nil {
return err
}
// Create each webhook
for _, hook := range runtimeListToUnstructured(mutHooks) {
log.V(1).Info("installing mutating webhook", "webhook", hook.GetName())
if err := ensureCreated(cs, hook); err != nil {
return err
}
}
for _, hook := range runtimeListToUnstructured(valHooks) {
log.V(1).Info("installing validating webhook", "webhook", hook.GetName())
if err := ensureCreated(cs, hook); err != nil {
return err
}
}
return nil
}
// ensureCreated creates or update object if already exists in the cluster
func ensureCreated(cs client.Client, obj *unstructured.Unstructured) error {
existing := obj.DeepCopy()
err := cs.Get(context.Background(), client.ObjectKey{Name: obj.GetName()}, existing)
switch {
case apierrors.IsNotFound(err):
if err := cs.Create(context.Background(), obj); err != nil {
return err
}
case err != nil:
return err
default:
log.V(1).Info("Webhook configuration already exists, updating", "webhook", obj.GetName())
obj.SetResourceVersion(existing.GetResourceVersion())
if err := cs.Update(context.Background(), obj); err != nil {
return err
}
}
return nil
}
// parseWebhookDirs reads the directories of Webhooks in options.DirectoryPaths and adds the Webhook structs to options
func parseWebhookDirs(options *WebhookInstallOptions) error {
if len(options.DirectoryPaths) > 0 {
for _, path := range options.DirectoryPaths {
_, err := os.Stat(path)
if options.IgnoreErrorIfPathMissing && os.IsNotExist(err) {
continue // skip this path
}
if !options.IgnoreErrorIfPathMissing && os.IsNotExist(err) {
return err // treat missing path as error
}
mutHooks, valHooks, err := readWebhooks(path)
if err != nil {
return err
}
options.MutatingWebhooks = append(options.MutatingWebhooks, mutHooks...)
options.ValidatingWebhooks = append(options.ValidatingWebhooks, valHooks...)
}
}
return nil
}
// readWebhooks reads the Webhooks from files and Unmarshals them into structs
// returns slice of mutating and validating webhook configurations
func readWebhooks(path string) ([]runtime.Object, []runtime.Object, error) {
// Get the webhook files
var files []os.FileInfo
var err error
log.V(1).Info("reading Webhooks from path", "path", path)
if files, err = ioutil.ReadDir(path); err != nil {
return nil, nil, err
}
// file extensions that may contain Webhooks
resourceExtensions := sets.NewString(".json", ".yaml", ".yml")
var mutHooks []runtime.Object
var valHooks []runtime.Object
for _, file := range files {
// Only parse whitelisted file types
if !resourceExtensions.Has(filepath.Ext(file.Name())) {
continue
}
// Unmarshal Webhooks from file into structs
docs, err := readDocuments(filepath.Join(path, file.Name()))
if err != nil {
return nil, nil, err
}
for _, doc := range docs {
var generic metav1.PartialObjectMetadata
if err = yaml.Unmarshal(doc, &generic); err != nil {
return nil, nil, err
}
const (
admissionregv1 = "admissionregistration.k8s.io/v1beta1"
admissionregv1beta1 = "admissionregistration.k8s.io/v1"
)
switch {
case generic.Kind == "MutatingWebhookConfiguration":
if generic.APIVersion != admissionregv1beta1 && generic.APIVersion != admissionregv1 {
return nil, nil, fmt.Errorf("only v1beta1 and v1 are supported right now for MutatingWebhookConfiguration (name: %s)", generic.Name)
}
hook := &unstructured.Unstructured{}
if err := yaml.Unmarshal(doc, &hook); err != nil {
return nil, nil, err
}
mutHooks = append(mutHooks, hook)
case generic.Kind == "ValidatingWebhookConfiguration":
if generic.APIVersion != admissionregv1beta1 && generic.APIVersion != admissionregv1 {
return nil, nil, fmt.Errorf("only v1beta1 and v1 are supported right now for ValidatingWebhookConfiguration (name: %s)", generic.Name)
}
hook := &unstructured.Unstructured{}
if err := yaml.Unmarshal(doc, &hook); err != nil {
return nil, nil, err
}
valHooks = append(valHooks, hook)
default:
continue
}
}
log.V(1).Info("read webhooks from file", "file", file.Name())
}
return mutHooks, valHooks, nil
}
func runtimeListToUnstructured(l []runtime.Object) []*unstructured.Unstructured {
res := []*unstructured.Unstructured{}
for _, obj := range l {
m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj.DeepCopyObject())
if err != nil {
continue
}
res = append(res, &unstructured.Unstructured{
Object: m,
})
}
return res
}

View file

@ -0,0 +1 @@
assets/bin

View file

@ -0,0 +1,10 @@
# Integration Testing Framework
This package has been moved from [https://github.com/kubernetes-sigs/testing_frameworks/tree/master/integration](https://github.com/kubernetes-sigs/testing_frameworks/tree/master/integration).
A framework for integration testing components of kubernetes. This framework is
intended to work properly both in CI, and on a local dev machine. It therefore
explicitly supports both Linux and Darwin.
For detailed documentation see the
[![GoDoc](https://godoc.org/github.com/kubernetes-sigs/controller-runtime/pkg/internal/testing/integration?status.svg)](https://godoc.org/github.com/kubernetes-sigs/controller-runtime/pkg/internal/testing/integration).

View file

@ -0,0 +1,74 @@
package addr
import (
"fmt"
"net"
"sync"
"time"
)
const (
portReserveTime = 1 * time.Minute
portConflictRetry = 100
)
type portCache struct {
lock sync.Mutex
ports map[int]time.Time
}
func (c *portCache) add(port int) bool {
c.lock.Lock()
defer c.lock.Unlock()
// remove outdated port
for p, t := range c.ports {
if time.Since(t) > portReserveTime {
delete(c.ports, p)
}
}
// try allocating new port
if _, ok := c.ports[port]; ok {
return false
}
c.ports[port] = time.Now()
return true
}
var cache = &portCache{
ports: make(map[int]time.Time),
}
func suggest() (port int, resolvedHost string, err error) {
addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
if err != nil {
return
}
l, err := net.ListenTCP("tcp", addr)
if err != nil {
return
}
port = l.Addr().(*net.TCPAddr).Port
defer func() {
err = l.Close()
}()
resolvedHost = addr.IP.String()
return
}
// Suggest suggests an address a process can listen on. It returns
// a tuple consisting of a free port and the hostname resolved to its IP.
// It makes sure that new port allocated does not conflict with old ports
// allocated within 1 minute.
func Suggest() (port int, resolvedHost string, err error) {
for i := 0; i < portConflictRetry; i++ {
port, resolvedHost, err = suggest()
if err != nil {
return
}
if cache.add(port) {
return
}
}
err = fmt.Errorf("no free ports found after %d retries", portConflictRetry)
return
}

View file

@ -0,0 +1,177 @@
package integration
import (
"fmt"
"io"
"io/ioutil"
"net/url"
"os"
"path/filepath"
"time"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/addr"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/internal"
)
// APIServer knows how to run a kubernetes apiserver.
type APIServer struct {
// URL is the address the ApiServer should listen on for client connections.
//
// If this is not specified, we default to a random free port on localhost.
URL *url.URL
// SecurePort is the additional secure port that the APIServer should listen on.
SecurePort int
// Path is the path to the apiserver binary.
//
// If this is left as the empty string, we will attempt to locate a binary,
// by checking for the TEST_ASSET_KUBE_APISERVER environment variable, and
// the default test assets directory. See the "Binaries" section above (in
// doc.go) for details.
Path string
// Args is a list of arguments which will passed to the APIServer binary.
// Before they are passed on, they will be evaluated as go-template strings.
// This means you can use fields which are defined and exported on this
// APIServer struct (e.g. "--cert-dir={{ .Dir }}").
// Those templates will be evaluated after the defaulting of the APIServer's
// fields has already happened and just before the binary actually gets
// started. Thus you have access to calculated fields like `URL` and others.
//
// If not specified, the minimal set of arguments to run the APIServer will
// be used.
Args []string
// CertDir is a path to a directory containing whatever certificates the
// APIServer will need.
//
// If left unspecified, then the Start() method will create a fresh temporary
// directory, and the Stop() method will clean it up.
CertDir string
// EtcdURL is the URL of the Etcd the APIServer should use.
//
// If this is not specified, the Start() method will return an error.
EtcdURL *url.URL
// StartTimeout, StopTimeout specify the time the APIServer is allowed to
// take when starting and stoppping before an error is emitted.
//
// If not specified, these default to 20 seconds.
StartTimeout time.Duration
StopTimeout time.Duration
// Out, Err specify where APIServer should write its StdOut, StdErr to.
//
// If not specified, the output will be discarded.
Out io.Writer
Err io.Writer
processState *internal.ProcessState
}
// Start starts the apiserver, waits for it to come up, and returns an error,
// if occurred.
func (s *APIServer) Start() error {
if s.processState == nil {
if err := s.setProcessState(); err != nil {
return err
}
}
return s.processState.Start(s.Out, s.Err)
}
func (s *APIServer) setProcessState() error {
if s.EtcdURL == nil {
return fmt.Errorf("expected EtcdURL to be configured")
}
var err error
s.processState = &internal.ProcessState{}
s.processState.DefaultedProcessInput, err = internal.DoDefaulting(
"kube-apiserver",
s.URL,
s.CertDir,
s.Path,
s.StartTimeout,
s.StopTimeout,
)
if err != nil {
return err
}
// Defaulting the secure port
if s.SecurePort == 0 {
s.SecurePort, _, err = addr.Suggest()
if err != nil {
return err
}
}
s.processState.HealthCheckEndpoint = "/healthz"
s.URL = &s.processState.URL
s.CertDir = s.processState.Dir
s.Path = s.processState.Path
s.StartTimeout = s.processState.StartTimeout
s.StopTimeout = s.processState.StopTimeout
if err := s.populateAPIServerCerts(); err != nil {
return err
}
s.processState.Args, err = internal.RenderTemplates(
internal.DoAPIServerArgDefaulting(s.Args), s,
)
return err
}
func (s *APIServer) populateAPIServerCerts() error {
_, statErr := os.Stat(filepath.Join(s.CertDir, "apiserver.crt"))
if !os.IsNotExist(statErr) {
return statErr
}
ca, err := internal.NewTinyCA()
if err != nil {
return err
}
certs, err := ca.NewServingCert()
if err != nil {
return err
}
certData, keyData, err := certs.AsBytes()
if err != nil {
return err
}
if err := ioutil.WriteFile(filepath.Join(s.CertDir, "apiserver.crt"), certData, 0640); err != nil {
return err
}
if err := ioutil.WriteFile(filepath.Join(s.CertDir, "apiserver.key"), keyData, 0640); err != nil {
return err
}
return nil
}
// Stop stops this process gracefully, waits for its termination, and cleans up
// the CertDir if necessary.
func (s *APIServer) Stop() error {
if s.processState != nil {
return s.processState.Stop()
}
return nil
}
// APIServerDefaultArgs exposes the default args for the APIServer so that you
// can use those to append your own additional arguments.
//
// The internal default arguments are explicitly copied here, we don't want to
// allow users to change the internal ones.
var APIServerDefaultArgs = append([]string{}, internal.APIServerDefaultArgs...)

View file

@ -0,0 +1,82 @@
package integration
import (
"fmt"
"net/url"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/internal"
)
// NewTinyCA creates a new a tiny CA utility for provisioning serving certs and client certs FOR TESTING ONLY.
// Don't use this for anything else!
var NewTinyCA = internal.NewTinyCA
// ControlPlane is a struct that knows how to start your test control plane.
//
// Right now, that means Etcd and your APIServer. This is likely to increase in
// future.
type ControlPlane struct {
APIServer *APIServer
Etcd *Etcd
}
// Start will start your control plane processes. To stop them, call Stop().
func (f *ControlPlane) Start() error {
if f.Etcd == nil {
f.Etcd = &Etcd{}
}
if err := f.Etcd.Start(); err != nil {
return err
}
if f.APIServer == nil {
f.APIServer = &APIServer{}
}
f.APIServer.EtcdURL = f.Etcd.URL
return f.APIServer.Start()
}
// Stop will stop your control plane processes, and clean up their data.
func (f *ControlPlane) Stop() error {
if f.APIServer != nil {
if err := f.APIServer.Stop(); err != nil {
return err
}
}
if f.Etcd != nil {
if err := f.Etcd.Stop(); err != nil {
return err
}
}
return nil
}
// APIURL returns the URL you should connect to to talk to your API.
func (f *ControlPlane) APIURL() *url.URL {
return f.APIServer.URL
}
// KubeCtl returns a pre-configured KubeCtl, ready to connect to this
// ControlPlane.
func (f *ControlPlane) KubeCtl() *KubeCtl {
k := &KubeCtl{}
k.Opts = append(k.Opts, fmt.Sprintf("--server=%s", f.APIURL()))
return k
}
// RESTClientConfig returns a pre-configured restconfig, ready to connect to
// this ControlPlane.
func (f *ControlPlane) RESTClientConfig() (*rest.Config, error) {
c := &rest.Config{
Host: f.APIURL().String(),
ContentConfig: rest.ContentConfig{
NegotiatedSerializer: serializer.WithoutConversionCodecFactory{CodecFactory: scheme.Codecs},
},
}
err := rest.SetKubernetesDefaults(c)
return c, err
}

View file

@ -0,0 +1,112 @@
/*
Package integration implements an integration testing framework for kubernetes.
It provides components for standing up a kubernetes API, against which you can test a
kubernetes client, or other kubernetes components. The lifecycle of the components
needed to provide this API is managed by this framework.
Quickstart
Add something like the following to
your tests:
cp := &integration.ControlPlane{}
cp.Start()
kubeCtl := cp.KubeCtl()
stdout, stderr, err := kubeCtl.Run("get", "pods")
// You can check on err, stdout & stderr and build up
// your tests
cp.Stop()
Components
Currently the framework provides the following components:
ControlPlane: The ControlPlane wraps Etcd & APIServer (see below) and wires
them together correctly. A ControlPlane can be stopped & started and can
provide the URL to connect to the API. The ControlPlane can also be asked for a
KubeCtl which is already correctly configured for this ControlPlane. The
ControlPlane is a good entry point for default setups.
Etcd: Manages an Etcd binary, which can be started, stopped and connected to.
By default Etcd will listen on a random port for http connections and will
create a temporary directory for its data. To configure it differently, see the
Etcd type documentation below.
APIServer: Manages an Kube-APIServer binary, which can be started, stopped and
connected to. By default APIServer will listen on a random port for http
connections and will create a temporary directory to store the (auto-generated)
certificates. To configure it differently, see the APIServer type
documentation below.
KubeCtl: Wraps around a `kubectl` binary and can `Run(...)` arbitrary commands
against a kubernetes control plane.
Binaries
Etcd, APIServer & KubeCtl use the same mechanism to determine which binaries to
use when they get started.
1. If the component is configured with a `Path` the framework tries to run that
binary.
For example:
myEtcd := &Etcd{
Path: "/some/other/etcd",
}
cp := &integration.ControlPlane{
Etcd: myEtcd,
}
cp.Start()
2. If the Path field on APIServer, Etcd or KubeCtl is left unset and an
environment variable named `TEST_ASSET_KUBE_APISERVER`, `TEST_ASSET_ETCD` or
`TEST_ASSET_KUBECTL` is set, its value is used as a path to the binary for the
APIServer, Etcd or KubeCtl.
3. If neither the `Path` field, nor the environment variable is set, the
framework tries to use the binaries `kube-apiserver`, `etcd` or `kubectl` in
the directory `${FRAMEWORK_DIR}/assets/bin/`.
Arguments for Etcd and APIServer
Those components will start without any configuration. However, if you want or
need to, you can override certain configuration -- one of which are the
arguments used when calling the binary.
When you choose to specify your own set of arguments, those won't be appended
to the default set of arguments, it is your responsibility to provide all the
arguments needed for the binary to start successfully.
However, the default arguments for APIServer and Etcd are exported as
`APIServerDefaultArgs` and `EtcdDefaultArgs` from this package. Treat those
variables as read-only constants. Internally we have a set of default
arguments for defaulting, the `APIServerDefaultArgs` and `EtcdDefaultArgs` are
just copies of those. So when you override them you loose access to the actual
internal default arguments, but your override won't affect the defaulting.
All arguments are interpreted as go templates. Those templates have access to
all exported fields of the `APIServer`/`Etcd` struct. It does not matter if
those fields where explicitly set up or if they were defaulted by calling the
`Start()` method, the template evaluation runs just before the binary is
executed and right after the defaulting of all the struct's fields has
happened.
// When you want to append additional arguments ...
etcd := &Etcd{
// Additional custom arguments will appended to the set of default
// arguments
Args: append(EtcdDefaultArgs, "--additional=arg"),
DataDir: "/my/special/data/dir",
}
// When you want to use a custom set of arguments ...
etcd := &Etcd{
// Only custom arguments will be passed to the binary
Args: []string{"--one=1", "--two=2", "--three=3"},
DataDir: "/my/special/data/dir",
}
*/
package integration

View file

@ -0,0 +1,114 @@
package integration
import (
"io"
"time"
"net/url"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/internal"
)
// Etcd knows how to run an etcd server.
type Etcd struct {
// URL is the address the Etcd should listen on for client connections.
//
// If this is not specified, we default to a random free port on localhost.
URL *url.URL
// Path is the path to the etcd binary.
//
// If this is left as the empty string, we will attempt to locate a binary,
// by checking for the TEST_ASSET_ETCD environment variable, and the default
// test assets directory. See the "Binaries" section above (in doc.go) for
// details.
Path string
// Args is a list of arguments which will passed to the Etcd binary. Before
// they are passed on, the`y will be evaluated as go-template strings. This
// means you can use fields which are defined and exported on this Etcd
// struct (e.g. "--data-dir={{ .Dir }}").
// Those templates will be evaluated after the defaulting of the Etcd's
// fields has already happened and just before the binary actually gets
// started. Thus you have access to calculated fields like `URL` and others.
//
// If not specified, the minimal set of arguments to run the Etcd will be
// used.
Args []string
// DataDir is a path to a directory in which etcd can store its state.
//
// If left unspecified, then the Start() method will create a fresh temporary
// directory, and the Stop() method will clean it up.
DataDir string
// StartTimeout, StopTimeout specify the time the Etcd is allowed to
// take when starting and stopping before an error is emitted.
//
// If not specified, these default to 20 seconds.
StartTimeout time.Duration
StopTimeout time.Duration
// Out, Err specify where Etcd should write its StdOut, StdErr to.
//
// If not specified, the output will be discarded.
Out io.Writer
Err io.Writer
processState *internal.ProcessState
}
// Start starts the etcd, waits for it to come up, and returns an error, if one
// occoured.
func (e *Etcd) Start() error {
if e.processState == nil {
if err := e.setProcessState(); err != nil {
return err
}
}
return e.processState.Start(e.Out, e.Err)
}
func (e *Etcd) setProcessState() error {
var err error
e.processState = &internal.ProcessState{}
e.processState.DefaultedProcessInput, err = internal.DoDefaulting(
"etcd",
e.URL,
e.DataDir,
e.Path,
e.StartTimeout,
e.StopTimeout,
)
if err != nil {
return err
}
e.processState.StartMessage = internal.GetEtcdStartMessage(e.processState.URL)
e.URL = &e.processState.URL
e.DataDir = e.processState.Dir
e.Path = e.processState.Path
e.StartTimeout = e.processState.StartTimeout
e.StopTimeout = e.processState.StopTimeout
e.processState.Args, err = internal.RenderTemplates(
internal.DoEtcdArgDefaulting(e.Args), e,
)
return err
}
// Stop stops this process gracefully, waits for its termination, and cleans up
// the DataDir if necessary.
func (e *Etcd) Stop() error {
return e.processState.Stop()
}
// EtcdDefaultArgs exposes the default args for Etcd so that you
// can use those to append your own additional arguments.
//
// The internal default arguments are explicitly copied here, we don't want to
// allow users to change the internal ones.
var EtcdDefaultArgs = append([]string{}, internal.EtcdDefaultArgs...)

View file

@ -0,0 +1,23 @@
package internal
var APIServerDefaultArgs = []string{
// Allow tests to run offline, by preventing API server from attempting to
// use default route to determine its --advertise-address
"--advertise-address=127.0.0.1",
"--etcd-servers={{ if .EtcdURL }}{{ .EtcdURL.String }}{{ end }}",
"--cert-dir={{ .CertDir }}",
"--insecure-port={{ if .URL }}{{ .URL.Port }}{{ end }}",
"--insecure-bind-address={{ if .URL }}{{ .URL.Hostname }}{{ end }}",
"--secure-port={{ if .SecurePort }}{{ .SecurePort }}{{ end }}",
"--disable-admission-plugins=NamespaceLifecycle,LimitRanger,ServiceAccount,TaintNodesByCondition,Priority,DefaultTolerationSeconds,DefaultStorageClass,StorageObjectInUseProtection,PersistentVolumeClaimResize,ResourceQuota", //nolint
"--service-cluster-ip-range=10.0.0.0/24",
"--allow-privileged=true",
}
func DoAPIServerArgDefaulting(args []string) []string {
if len(args) != 0 {
return args
}
return APIServerDefaultArgs
}

View file

@ -0,0 +1,28 @@
package internal
import (
"bytes"
"html/template"
)
func RenderTemplates(argTemplates []string, data interface{}) (args []string, err error) {
var t *template.Template
for _, arg := range argTemplates {
t, err = template.New(arg).Parse(arg)
if err != nil {
args = nil
return
}
buf := &bytes.Buffer{}
err = t.Execute(buf, data)
if err != nil {
args = nil
return
}
args = append(args, buf.String())
}
return
}

View file

@ -0,0 +1,35 @@
package internal
import (
"os"
"path/filepath"
"regexp"
"runtime"
"strings"
)
var assetsPath string
func init() {
_, thisFile, _, ok := runtime.Caller(0)
if !ok {
panic("Could not determine the path of the BinPathFinder")
}
assetsPath = filepath.Join(filepath.Dir(thisFile), "..", "assets", "bin")
}
// BinPathFinder checks the an environment variable, derived from the symbolic name,
// and falls back to a default assets location when this variable is not set
func BinPathFinder(symbolicName string) (binPath string) {
punctuationPattern := regexp.MustCompile("[^A-Z0-9]+")
sanitizedName := punctuationPattern.ReplaceAllString(strings.ToUpper(symbolicName), "_")
leadingNumberPattern := regexp.MustCompile("^[0-9]+")
sanitizedName = leadingNumberPattern.ReplaceAllString(sanitizedName, "")
envVar := "TEST_ASSET_" + sanitizedName
if val, ok := os.LookupEnv(envVar); ok {
return val
}
return filepath.Join(assetsPath, symbolicName)
}

View file

@ -0,0 +1,38 @@
package internal
import (
"net/url"
)
var EtcdDefaultArgs = []string{
"--listen-peer-urls=http://localhost:0",
"--advertise-client-urls={{ if .URL }}{{ .URL.String }}{{ end }}",
"--listen-client-urls={{ if .URL }}{{ .URL.String }}{{ end }}",
"--data-dir={{ .DataDir }}",
}
func DoEtcdArgDefaulting(args []string) []string {
if len(args) != 0 {
return args
}
return EtcdDefaultArgs
}
func isSecureScheme(scheme string) bool {
// https://github.com/coreos/etcd/blob/d9deeff49a080a88c982d328ad9d33f26d1ad7b6/pkg/transport/listener.go#L53
if scheme == "https" || scheme == "unixs" {
return true
}
return false
}
func GetEtcdStartMessage(listenURL url.URL) string {
if isSecureScheme(listenURL.Scheme) {
// https://github.com/coreos/etcd/blob/a7f1fbe00ec216fcb3a1919397a103b41dca8413/embed/serve.go#L167
return "serving client requests on "
}
// https://github.com/coreos/etcd/blob/a7f1fbe00ec216fcb3a1919397a103b41dca8413/embed/serve.go#L124
return "serving insecure client requests on "
}

View file

@ -0,0 +1,217 @@
package internal
import (
"fmt"
"io"
"io/ioutil"
"net"
"net/http"
"net/url"
"os"
"os/exec"
"path"
"strconv"
"time"
"github.com/onsi/gomega/gbytes"
"github.com/onsi/gomega/gexec"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/addr"
)
type ProcessState struct {
DefaultedProcessInput
Session *gexec.Session
// Healthcheck Endpoint. If we get http.StatusOK from this endpoint, we
// assume the process is ready to operate. E.g. "/healthz". If this is set,
// we ignore StartMessage.
HealthCheckEndpoint string
// HealthCheckPollInterval is the interval which will be used for polling the
// HealthCheckEndpoint.
// If left empty it will default to 100 Milliseconds.
HealthCheckPollInterval time.Duration
// StartMessage is the message to wait for on stderr. If we receive this
// message, we assume the process is ready to operate. Ignored if
// HealthCheckEndpoint is specified.
//
// The usage of StartMessage is discouraged, favour HealthCheckEndpoint
// instead!
//
// Deprecated: Use HealthCheckEndpoint in favour of StartMessage
StartMessage string
Args []string
// ready holds wether the process is currently in ready state (hit the ready condition) or not.
// It will be set to true on a successful `Start()` and set to false on a successful `Stop()`
ready bool
}
type DefaultedProcessInput struct {
URL url.URL
Dir string
DirNeedsCleaning bool
Path string
StopTimeout time.Duration
StartTimeout time.Duration
}
func DoDefaulting(
name string,
listenURL *url.URL,
dir string,
path string,
startTimeout time.Duration,
stopTimeout time.Duration,
) (DefaultedProcessInput, error) {
defaults := DefaultedProcessInput{
Dir: dir,
Path: path,
StartTimeout: startTimeout,
StopTimeout: stopTimeout,
}
if listenURL == nil {
port, host, err := addr.Suggest()
if err != nil {
return DefaultedProcessInput{}, err
}
defaults.URL = url.URL{
Scheme: "http",
Host: net.JoinHostPort(host, strconv.Itoa(port)),
}
} else {
defaults.URL = *listenURL
}
if dir == "" {
newDir, err := ioutil.TempDir("", "k8s_test_framework_")
if err != nil {
return DefaultedProcessInput{}, err
}
defaults.Dir = newDir
defaults.DirNeedsCleaning = true
}
if path == "" {
if name == "" {
return DefaultedProcessInput{}, fmt.Errorf("must have at least one of name or path")
}
defaults.Path = BinPathFinder(name)
}
if startTimeout == 0 {
defaults.StartTimeout = 20 * time.Second
}
if stopTimeout == 0 {
defaults.StopTimeout = 20 * time.Second
}
return defaults, nil
}
type stopChannel chan struct{}
func (ps *ProcessState) Start(stdout, stderr io.Writer) (err error) {
if ps.ready {
return nil
}
command := exec.Command(ps.Path, ps.Args...)
ready := make(chan bool)
timedOut := time.After(ps.StartTimeout)
var pollerStopCh stopChannel
if ps.HealthCheckEndpoint != "" {
healthCheckURL := ps.URL
healthCheckURL.Path = ps.HealthCheckEndpoint
pollerStopCh = make(stopChannel)
go pollURLUntilOK(healthCheckURL, ps.HealthCheckPollInterval, ready, pollerStopCh)
} else {
startDetectStream := gbytes.NewBuffer()
ready = startDetectStream.Detect(ps.StartMessage)
stderr = safeMultiWriter(stderr, startDetectStream)
}
ps.Session, err = gexec.Start(command, stdout, stderr)
if err != nil {
return err
}
select {
case <-ready:
ps.ready = true
return nil
case <-timedOut:
if pollerStopCh != nil {
close(pollerStopCh)
}
if ps.Session != nil {
ps.Session.Terminate()
}
return fmt.Errorf("timeout waiting for process %s to start", path.Base(ps.Path))
}
}
func safeMultiWriter(writers ...io.Writer) io.Writer {
safeWriters := []io.Writer{}
for _, w := range writers {
if w != nil {
safeWriters = append(safeWriters, w)
}
}
return io.MultiWriter(safeWriters...)
}
func pollURLUntilOK(url url.URL, interval time.Duration, ready chan bool, stopCh stopChannel) {
if interval <= 0 {
interval = 100 * time.Millisecond
}
for {
res, err := http.Get(url.String())
if err == nil {
res.Body.Close()
if res.StatusCode == http.StatusOK {
ready <- true
return
}
}
select {
case <-stopCh:
return
default:
time.Sleep(interval)
}
}
}
func (ps *ProcessState) Stop() error {
if ps.Session == nil {
return nil
}
// gexec's Session methods (Signal, Kill, ...) do not check if the Process is
// nil, so we are doing this here for now.
// This should probably be fixed in gexec.
if ps.Session.Command.Process == nil {
return nil
}
detectedStop := ps.Session.Terminate().Exited
timedOut := time.After(ps.StopTimeout)
select {
case <-detectedStop:
break
case <-timedOut:
return fmt.Errorf("timeout waiting for process %s to stop", path.Base(ps.Path))
}
ps.ready = false
if ps.DirNeedsCleaning {
return os.RemoveAll(ps.Dir)
}
return nil
}

View file

@ -0,0 +1,149 @@
package internal
// NB(directxman12): nothing has verified that this has good settings. In fact,
// the setting generated here are probably terrible, but they're fine for integration
// tests. These ABSOLUTELY SHOULD NOT ever be exposed in the public API. They're
// ONLY for use with envtest's ability to configure webhook testing.
// If I didn't otherwise not want to add a dependency on cfssl, I'd just use that.
import (
"crypto"
crand "crypto/rand"
"crypto/rsa"
"crypto/x509"
"crypto/x509/pkix"
"encoding/pem"
"fmt"
"math/big"
"net"
"time"
certutil "k8s.io/client-go/util/cert"
)
var (
rsaKeySize = 2048 // a decent number, as of 2019
bigOne = big.NewInt(1)
)
// CertPair is a private key and certificate for use for client auth, as a CA, or serving.
type CertPair struct {
Key crypto.Signer
Cert *x509.Certificate
}
// CertBytes returns the PEM-encoded version of the certificate for this pair.
func (k CertPair) CertBytes() []byte {
return pem.EncodeToMemory(&pem.Block{
Type: "CERTIFICATE",
Bytes: k.Cert.Raw,
})
}
// AsBytes encodes keypair in the appropriate formats for on-disk storage (PEM and
// PKCS8, respectively).
func (k CertPair) AsBytes() (cert []byte, key []byte, err error) {
cert = k.CertBytes()
rawKeyData, err := x509.MarshalPKCS8PrivateKey(k.Key)
if err != nil {
return nil, nil, fmt.Errorf("unable to encode private key: %v", err)
}
key = pem.EncodeToMemory(&pem.Block{
Type: "PRIVATE KEY",
Bytes: rawKeyData,
})
return cert, key, nil
}
// TinyCA supports signing serving certs and client-certs,
// and can be used as an auth mechanism with envtest.
type TinyCA struct {
CA CertPair
orgName string
nextSerial *big.Int
}
// newPrivateKey generates a new private key of a relatively sane size (see
// rsaKeySize).
func newPrivateKey() (crypto.Signer, error) {
return rsa.GenerateKey(crand.Reader, rsaKeySize)
}
func NewTinyCA() (*TinyCA, error) {
caPrivateKey, err := newPrivateKey()
if err != nil {
return nil, fmt.Errorf("unable to generate private key for CA: %v", err)
}
caCfg := certutil.Config{CommonName: "envtest-environment", Organization: []string{"envtest"}}
caCert, err := certutil.NewSelfSignedCACert(caCfg, caPrivateKey)
if err != nil {
return nil, fmt.Errorf("unable to generate certificate for CA: %v", err)
}
return &TinyCA{
CA: CertPair{Key: caPrivateKey, Cert: caCert},
orgName: "envtest",
nextSerial: big.NewInt(1),
}, nil
}
func (c *TinyCA) makeCert(cfg certutil.Config) (CertPair, error) {
now := time.Now()
key, err := newPrivateKey()
if err != nil {
return CertPair{}, fmt.Errorf("unable to create private key: %v", err)
}
serial := new(big.Int).Set(c.nextSerial)
c.nextSerial.Add(c.nextSerial, bigOne)
template := x509.Certificate{
Subject: pkix.Name{CommonName: cfg.CommonName, Organization: cfg.Organization},
DNSNames: cfg.AltNames.DNSNames,
IPAddresses: cfg.AltNames.IPs,
SerialNumber: serial,
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
ExtKeyUsage: cfg.Usages,
// technically not necessary for testing, but let's set anyway just in case.
NotBefore: now.UTC(),
// 1 week -- the default for cfssl, and just long enough for a
// long-term test, but not too long that anyone would try to use this
// seriously.
NotAfter: now.Add(168 * time.Hour).UTC(),
}
certRaw, err := x509.CreateCertificate(crand.Reader, &template, c.CA.Cert, key.Public(), c.CA.Key)
if err != nil {
return CertPair{}, fmt.Errorf("unable to create certificate: %v", err)
}
cert, err := x509.ParseCertificate(certRaw)
if err != nil {
return CertPair{}, fmt.Errorf("generated invalid certificate, could not parse: %v", err)
}
return CertPair{
Key: key,
Cert: cert,
}, nil
}
// NewServingCert returns a new CertPair for a serving HTTPS on localhost.
func (c *TinyCA) NewServingCert() (CertPair, error) {
return c.makeCert(certutil.Config{
CommonName: "localhost",
Organization: []string{c.orgName},
AltNames: certutil.AltNames{
DNSNames: []string{"localhost"},
IPs: []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
},
Usages: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
})
}

View file

@ -0,0 +1,47 @@
package integration
import (
"bytes"
"io"
"os/exec"
"sigs.k8s.io/controller-runtime/pkg/internal/testing/integration/internal"
)
// KubeCtl is a wrapper around the kubectl binary.
type KubeCtl struct {
// Path where the kubectl binary can be found.
//
// If this is left empty, we will attempt to locate a binary, by checking for
// the TEST_ASSET_KUBECTL environment variable, and the default test assets
// directory. See the "Binaries" section above (in doc.go) for details.
Path string
// Opts can be used to configure additional flags which will be used each
// time the wrapped binary is called.
//
// For example, you might want to use this to set the URL of the APIServer to
// connect to.
Opts []string
}
// Run executes the wrapped binary with some preconfigured options and the
// arguments given to this method. It returns Readers for the stdout and
// stderr.
func (k *KubeCtl) Run(args ...string) (stdout, stderr io.Reader, err error) {
if k.Path == "" {
k.Path = internal.BinPathFinder("kubectl")
}
stdoutBuffer := &bytes.Buffer{}
stderrBuffer := &bytes.Buffer{}
allArgs := append(k.Opts, args...)
cmd := exec.Command(k.Path, allArgs...)
cmd.Stdout = stdoutBuffer
cmd.Stderr = stderrBuffer
err = cmd.Run()
return stdoutBuffer, stderrBuffer, err
}