Life of a Kubernetes API Request

Slides for talk given at Google Open Source Live - Kubernetes Day on 2020-12-03

https://www.youtube.com/GoogleOpenSource

Transcript

1. google_l Open Source Life of a Kubernetes API Request Kevin Delgado Software Engineer
2. google_l Open Source What happens when you type $ kubectl create -f rs.yaml
3. google_l Open Source The kubectl client CLI Setup 1 Command Execution 2 HTTP Request Building 3 Serialization, Conversion, Defaulting 4
4. google_l Open Source CLI setup Build all the cobra commands… …with a Factory // kubectl/pkg/cmd/create/create.go func NewCmdCreate(f cmdutil.Factory, ioStreams genericclioptions.IOStreams) *cobra.Command { o := NewCreateOptions(ioStreams) cmd := &cobra.Command{ … // documentation setup Run: func(cmd *cobra.Command, args []string) { … // error handling and validation cmdutil.CheckErr(o.RunCreate(f, cmd)) }, } … // add subcommands return cmd } // kubectl/pkg/cmd/util/factory.go type Factory interface { genericclioptions.RESTClientGetter // Returns a RESTClient for accessing Kubernetes resources or an error. RESTClient() (*restclient.RESTClient, error) // NewBuilder returns an object that assists in loading objects from both disk and the server // and which implements the common patterns for CLI interactions with generic resources. NewBuilder() *resource.Builder // Returns a schema that can validate objects stored on disk. Validator(validate bool) (validation.Schema, error) // OpenAPISchema returns the schema openapi schema definition OpenAPISchema() (openapi.Resources, error) … // other clients (DynamicClient, KubernetesClietSet, etc) }
5. google_l Open Source Command Execution Builder iterates through each resource... …using Helper to call generic REST verb // kubectl/pkg/cmd/create/create.go func (o *CreateOptions) RunCreate(f cmdutil.Factory, cmd *cobra.Command) error { … // setup r := f.NewBuilder(). Unstructured(). … // other builder options Do() err = r.Visit(func(info *resource.Info, err error) error { … // annotations, recording, dry run strategy, etc.. obj, err := resource. NewHelper(info.Client, info.Mapping). … // other helper options Create(info.Namespace, true, info.Object) … // handle and return any errors }) } // cli-runtime/pkg/resource/helper.go func (m *Helper) Create(namespace string, modify bool, obj runtime.Object, options *metav1.CreateOptions) (runtime.Object, error) { … // deal with options and object versioning return m.createResource(m.RESTClient, m.Resource, namespace, obj, options) } func (m *Helper) createResource(c RESTClient, resource, namespace string, obj runtime.Object, options *metav1.CreateOptions) (runtime.Object, error) { return c.Post(). NamespaceIfScoped(namespace, m.NamespaceScoped). Resource(resource). VersionedParams(options, metav1.ParameterCodec). Body(obj). Do(context.TODO()). Get() }
6. google_l Open Source HTTP Request Building & Request Execution REST Client invokes an HTTP Post... …on the Request and Body it builds // client-go/rest/client.go func (c *RESTClient) Post() *Request { return c.Verb("POST") } func (c *RESTClient) Verb(verb string) *Request { return NewRequest(c).Verb(verb) } // client-go/rest/request.go func (r *Request) Body(obj interface{}) *Request { switch t := obj.(type) { case string: r.body = bytes.NewReader(data) case []byte: r.body = bytes.NewReader(t) case io.Reader: r.body = t case runtime.Object: encoder, err := r.c.content.Negotiator.Encoder(r.c.content.ContentType, nil) data, err := runtime.Encode(encoder, t) r.body = bytes.NewReader(data) default: r.err = fmt.Errorf("unknown type used for body: %+v", obj) } return r } // client-go/rest/request.go func (r *Request) Do(ctx context.Context) Result { var result Result err := r.request(ctx, func(req *http.Request, resp *http.Response) { result = r.transformResponse(resp, req) }) return result }
7. google_l Open Source Serialization, Conversion, Defaulting Prior to executing the Request, the client serializes the body into a wire format // apimachinery/pkg/runtime/interfaces.go type Encoder interface { Encode(obj Object, w io.Writer) error Identifier() Identifier } type Decoder interface { Decode(data []byte, defaults *schema.GroupVersionKind, into Object) (Object, *schema.GroupVersionKind, error) } type Object interface { GetObjectKind() schema.ObjectKind DeepCopyObject() Object } // apimachinery/pkg/runtime/interfaces.go type Serializer interface { Encoder Decoder } type NegotiatedSerializer interface SupportedMediaTypes() []SerializerInfo EncoderForVersion(serializer Encoder, gv GroupVersioner) Encoder DecoderToVersion(serializer Decoder, gv GroupVersioner) Decoder } type Codec Serializer
8. google_l Open Source The kube-api-server Server Aggregation Layer Generic API Server Setup Routing & Dispatch Request Handling 1 2 3 4 5 Storage to etcd
9. google_l Open Source Aggregation Layer Run the main method… // kubernetes/cmd/kube-apiserver/apiserver.go func main() { command := app.NewAPIServerCommand() if err := command.Execute(); err != nil { os.Exit(1) } } // kubernetes/cmd/kube-apiserver/app/server.go func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan struct{}) (*aggregatorapiserver.APIAggregator, error) { … // create configs for extensions and kube api servers. apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegate()) kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer) aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers) return aggregatorServer, nil } func CreateKubeAPIServer(kubeAPIServerConfig *controlplane.Config, delegateAPIServer genericapiserver.DelegationTarget) (*controlplane.Instance, error) { kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer) return kubeAPIServer, nil } …and aggregate the API servers // kubernetes/cmd/kube-apiserver/app/server.go func NewAPIServerCommand() *cobra.Command { cmd := &cobra.Command{ RunE: func(cmd *cobra.Command, args []string) error { return Run(completedOptions, genericapiserver.SetupSignalHandler()) },} return cmd } func Run(completeOptions completedServerRunOptions, stopCh <-chan struct{}) error { server, err := CreateServerChain(completeOptions, stopCh) prepared, err := server.PrepareRun() return prepared.Run(stopCh) }
10. google_l Open Source API Server Setup Create the GenericAPIServer... // apiserver/pkg/server/config.go func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*GenericAPIServer, error) { handlerChainBuilder := func(handler http.Handler) http.Handler { return c.BuildHandlerChainFunc(handler, c.Config) } apiServerHandler := NewAPIServerHandler(name, c.Serializer, handlerChainBuilder, delegationTarget.UnprotectedHandler()) s := &GenericAPIServer{ … // add all the state for the GenericAPIServer, for example: Serializer: c.Serializer, } … // add hooks (poststart, priority/fairness) and healthchecks installAPI(s, c.Config) … // return the generic api server any errors } // apiserver/pkg/server/config.go func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler { handler := genericapifilters.WithAuthorization(apiHandler, c.Authorization.Authorizer, c.Serializer) handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true") handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc, c.RequestTimeout) handler = genericapifilters.WithWarningRecorder(handler) handler = genericapifilters.WithCacheControl(handler) handler = genericfilters.WithPanicRecovery(handler) … // and many more... return handler } … and build the handler chain
11. google_l Open Source API Server Setup (cont.) Run the GenericAPIServer... // apiserver/pkg/server/genericapiserver.go func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error { … // ensure server shuts down gracefully after a built in delay stoppedCh, err := s.NonBlockingRun(delayedStopCh) … // run shutdown hooks, wait for requests to finish, return errors, etc } func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}) (<-chan struct{}, error) { … // start audit backend if applicable, if s.SecureServingInfo != nil && s.Handler != nil { stoppedCh, err = s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh) } … // clean up stop channels, run post stop hooks, return errors, etc } // apiserver/pkg/server/secure_serving.go func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Duration, stopCh <-chan struct{}) (<-chan struct{}, error) { … // Set up TLS, http2 options, etc secureServer := &http.Server{ Addr: s.Listener.Addr().String(), Handler: handler, MaxHeaderBytes: 1 << 20, TLSConfig: tlsConfig, } return RunServer(secureServer, s.Listener, shutdownTimeout, stopCh) } func RunServer(server *http.Server, ln net.Listener, shutDownTimeout time.Duration, stopCh <-chan struct{}) (<-chan struct{}, error) { … // deal with server shutdown go func() { … // set up tcpKeepAliveListener with TLS, etc err := server.Serve(listener) }() … // return stoppedCh and any errors } … and start the underlying http Server
12. google_l Open Source Routing & Dispatch (go-restful muxer) Using the APIGroupVersion, the Installer sets up dispatching for specific resources // apiserver/pkg/endpoints/groupversion.go type APIGroupVersion struct { Storage map[string]rest.Storage GroupVersion schema.GroupVersion Serializer runtime.NegotiatedSerializer ParameterCodec runtime.ParameterCodec Creater runtime.ObjectCreater Convertor runtime.ObjectConvertor Defaulter runtime.ObjectDefaulter … // more fields } // apiserver/pkg/endpoints/installer.go func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService) (*metav1.APIResource, error) { creater, isCreater := storage.(rest.Creater) for _, action := range actions { switch action.Verb { … // cases for all HTTP verbs case "POST": // Create a resource. … // create the handler route := ws.POST(action.Path).To(handler). Doc(doc). … // more functions to build the route Returns(http.StatusOK, "OK", producedObject). Writes(producedObject) routes = append(routes, route) return &apiResource, nil } // apiserver/pkg/endpoints/installer.go func (a *APIInstaller) Install() ([]metav1.APIResource, *restful.WebService, []error) { paths := make([]string, len(a.group.Storage)) for _, path := range paths { apiResource, err := a.registerResourceHandlers(path, a.group.Storage[path], ws) } return apiResources, ws, errors }
13. google_l Open Source Request Handling CreateResource handles the HTTP request… …including deserialization, conversion, and defaulting… …and writing out to storage // apiserver/pkg/endpoints/handlers/create.go func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Interface, includeName bool) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { … // deal with tracing, dry-run, namespace, timouts decoder := scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion) body, err := limitedReadBody(req, scope.MaxRequestBodyBytes) … // retrieve query params and validate create options obj, gvk, err := decoder.Decode(body, &defaultGVK, original) admissionAttributes := admission.NewAttributesRecord(obj, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Create, options, dryrun.IsDryRun(options.DryRun), userInfo) requestFunc := func() (runtime.Object, error) { return r.Create(ctx, name, obj, rest.AdmissionToValidateObjectFunc(admit, admissionAttributes, scope), options, ) } result, err := finishRequest(timeout, func() (runtime.Object, error) { if mutatingAdmission, ok := admit.(admission.MutationInterface); ok && mutatingAdmission.Handles(admission.Create) { if err := mutatingAdmission.Admit(ctx, admissionAttributes, scope); err != nil { return nil, err } } result, err := requestFunc() return result, err }) code := http.StatusCreated transformResponseObject(ctx, scope, trace, req, w, code, outputMediaType, result) } }
14. google_l Open Source Storage to etcd Storage Interfaces 1 Storage Strategy 2 etcd3 Client 3
15. google_l Open Source Storage Interfaces Invoke the Creater interface... … that is implemented by Store // apiserver/pkg/registry/rest/rest.go type NamedCreater interface { New() runtime.Object Create(ctx context.Context, name string, obj runtime.Object, createValidation ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) } // apiserver/pkg/registry/generic/registry/store.go func (e *Store) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) { rest.BeforeCreate(e.CreateStrategy, ctx, obj) … // call validators, retrieve object’s name key, err := e.KeyFunc(ctx, name) ttl, err := e.calculateTTL(obj, 0, false) out := e.NewFunc() if err := e.Storage.Create(ctx, key, obj, out, ttl, dryrun.IsDryRun(options.DryRun)); err != nil { … // handle errors from writing to storage } e.AfterCreate(out) … // apply decorator, handle errors return out, nil } // apiserver/pkg/registry/genericy/registry/store.go type Store struct { NewFunc func() runtime.Object KeyFunc func(ctx context.Context, name string) (string, error) ObjectNameFunc func(obj runtime.Object) (string, error) CreateStrategy rest.RESTCreateStrategy AfterCreate ObjectFunc Storage DryRunnableStorage … // more fields }
16. google_l Open Source Storage Strategy Use the RESTCreateStrategy specific to replicasets // apiserver/pkg/registry/rest/create.go type RESTCreateStrategy interface { runtime.ObjectTyper names.NameGenerator NamespaceScoped() bool PrepareForCreate(ctx context.Context, obj runtime.Object) Validate(ctx context.Context, obj runtime.Object) field.ErrorList Canonicalize(obj runtime.Object) } // kubernetes/pkg/registry/apps/replicaset/strategy.go type rsStrategy struct { runtime.ObjectTyper names.NameGenerator } func (rsStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) { rs := obj.(*apps.ReplicaSet) rs.Status = apps.ReplicaSetStatus{} rs.Generation = 1 pod.DropDisabledTemplateFields(&rs.Spec.Template, nil) } func (rsStrategy) Validate(ctx context.Context, obj runtime.Object) field.ErrorList { rs := obj.(*apps.ReplicaSet) allErrs := validation.ValidateReplicaSet(rs) allErrs = append(allErrs, corevalidation.ValidateConditionalPodTemplate(&rs.Spec.Template, nil, field.NewPath("spec.template"))...) return allErrs }
17. google_l Open Source Storage Strategy (cont.) During api setup, install the replicaset storage provider // kubernetes/pkg/controlplane/instance.go (master/master.go previously) type Instance struct { GenericAPIServer *genericapiserver.GenericAPIServer ClusterAuthenticationInfo clusterauthenticationtrust.ClusterAuthenticationInfo } func (m *Instance) InstallAPIs(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter, restStorageProviders ...RESTStorageProvider) error { apiGroupsInfo := []*genericapiserver.APIGroupInfo{} for _, restStorageBuilder := range restStorageProviders { groupName := restStorageBuilder.GroupName() apiGroupInfo, enabled, err := restStorageBuilder.NewRESTStorage(apiResourceConfigSource, restOptionsGetter) .. // handle errors and and add post start hooks apiGroupsInfo = append(apiGroupsInfo, &apiGroupInfo) } m.GenericAPIServer.InstallAPIGroups(apiGroupsInfo...) return nil } // kubernetes/pkg/registry/apps/replicaset/storage/storage.go func NewStorage(optsGetter generic.RESTOptionsGetter) (ReplicaSetStorage, error) { replicaSetRest, replicaSetStatusRest, err := NewREST(optsGetter) return ReplicaSetStorage{ ReplicaSet: replicaSetRest, Status: replicaSetStatusRest, Scale: &ScaleREST{store: replicaSetRest.Store}, }, nil } func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST, error) { store := &genericregistry.Store{ NewFunc: func() runtime.Object { return &apps.ReplicaSet{} }, CreateStrategy: replicaset.Strategy, … // other fields, functions, and strategies on the store } options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: replicaset.GetAttrs} store.CompleteWithOptions(options) return &REST{store, []string{"all"}}, &StatusREST{store: &statusStore}, nil }
18. google_l Open Source etcd3 Client Finally it uses the etcd3 client to store to etcd // apiserver/pkg/storage/etcd3/store.go func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error { version, err := s.versioner.ObjectResourceVersion(obj) s.versioner.PrepareObjectForStorage(obj); data, err := runtime.Encode(s.codec, obj) key = path.Join(s.pathPrefix, key) opts, err := s.ttlOpts(ctx, int64(ttl)) newData, err := s.transformer.TransformToStorage(data, authenticatedDataString(key)) startTime := time.Now() txnResp, err := s.client.KV.Txn(ctx).If( notFound(key), ).Then( clientv3.OpPut(key, string(newData), opts...), ).Commit() if out != nil { putResp := txnResp.Responses[0].GetResponsePut() return decode(s.codec, s.versioner, data, out, putResp.Header.Revision) } return nil }
19. google_l Open Source Thank You!
20. Thank you! Kevin Delgado Software Engineer GitHub: @kevindelgado