如何初始化k8s中的client
1、kubernetes.Clientset 参考链接
- 集群内访问创建k8s-client - 直接获取集群内的config, 通过config创建clientSet。
// creates the in-cluster config
config, err := rest.InClusterConfig()
if err != nil {
panic(err.Error())
}
// creates the clientset
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err.Error())
}
// InClusterConfig returns a config object which uses the service account
// kubernetes gives to pods. It's intended for clients that expect to be
// running inside a pod running on kubernetes. It will return ErrNotInCluster
// if called from a process not running in a kubernetes environment.
func InClusterConfig() (*Config, error) {
const (
tokenFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
rootCAFile = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
)
host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
if len(host) == 0 || len(port) == 0 {
return nil, ErrNotInCluster
}
token, err := ioutil.ReadFile(tokenFile)
if err != nil {
return nil, err
}
tlsClientConfig := TLSClientConfig{}
if _, err := certutil.NewPool(rootCAFile); err != nil {
klog.Errorf("Expected to load root CA config from %s, but got err: %v", rootCAFile, err)
} else {
tlsClientConfig.CAFile = rootCAFile
}
return &Config{
// TODO: switch to using cluster DNS.
Host: "https://" + net.JoinHostPort(host, port),
TLSClientConfig: tlsClientConfig,
BearerToken: string(token),
BearerTokenFile: tokenFile,
}, nil
}
- 集群外访问创建k8s-client，由于client是在集群外，与集群内的client相比要多考虑如何授权集群外的k8s-api访问的逻辑。我们可以使用包含集群上下文信息的kubeconfig文件来初始化客户端。 kubectl命令也使用kubeconfig文件对集群进行身份验证。
var kubeconfig *string
if home := homeDir(); home != "" {
kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
} else {
kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
}
flag.Parse()
// use the current context in kubeconfig
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
if err != nil {
panic(err.Error())
}
// create the clientset
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err.Error())
}
- 使用kubeconfig文件来组织关于集群，用户，名称空间和身份验证机制的信息。 kubectl命令行工具使用kubeconfig文件来查找选择群集并与群集的API服务器进行通信所需的信息。默认情况下 kubectl使用的配置文件名称是在$HOME/.kube目录下 config文件，可以通过设置环境变量KUBECONFIG或者–kubeconfig指定其他的配置文件
2、RESTClient
k8s-clientSet是各个k8s资源的客户端集合，先看下k8s-clientSet的结构体，结构体中每个成员实质上为k8s每个资源对应的restful-client。
// Clientset contains the clients for groups. Each group has exactly one
// version included in a Clientset.
type Clientset struct {
*discovery.DiscoveryClient
admissionregistrationV1 *admissionregistrationv1.AdmissionregistrationV1Client
admissionregistrationV1beta1 *admissionregistrationv1beta1.AdmissionregistrationV1beta1Client
appsV1 *appsv1.AppsV1Client
appsV1beta1 *appsv1beta1.AppsV1beta1Client
appsV1beta2 *appsv1beta2.AppsV1beta2Client
auditregistrationV1alpha1 *auditregistrationv1alpha1.AuditregistrationV1alpha1Client
authenticationV1 *authenticationv1.AuthenticationV1Client
authenticationV1beta1 *authenticationv1beta1.AuthenticationV1beta1Client
authorizationV1 *authorizationv1.AuthorizationV1Client
authorizationV1beta1 *authorizationv1beta1.AuthorizationV1beta1Client
autoscalingV1 *autoscalingv1.AutoscalingV1Client
autoscalingV2beta1 *autoscalingv2beta1.AutoscalingV2beta1Client
autoscalingV2beta2 *autoscalingv2beta2.AutoscalingV2beta2Client
batchV1 *batchv1.BatchV1Client
batchV1beta1 *batchv1beta1.BatchV1beta1Client
batchV2alpha1 *batchv2alpha1.BatchV2alpha1Client
certificatesV1beta1 *certificatesv1beta1.CertificatesV1beta1Client
coordinationV1beta1 *coordinationv1beta1.CoordinationV1beta1Client
coordinationV1 *coordinationv1.CoordinationV1Client
coreV1 *corev1.CoreV1Client
discoveryV1alpha1 *discoveryv1alpha1.DiscoveryV1alpha1Client
discoveryV1beta1 *discoveryv1beta1.DiscoveryV1beta1Client
eventsV1beta1 *eventsv1beta1.EventsV1beta1Client
extensionsV1beta1 *extensionsv1beta1.ExtensionsV1beta1Client
flowcontrolV1alpha1 *flowcontrolv1alpha1.FlowcontrolV1alpha1Client
networkingV1 *networkingv1.NetworkingV1Client
networkingV1beta1 *networkingv1beta1.NetworkingV1beta1Client
nodeV1alpha1 *nodev1alpha1.NodeV1alpha1Client
nodeV1beta1 *nodev1beta1.NodeV1beta1Client
policyV1beta1 *policyv1beta1.PolicyV1beta1Client
rbacV1 *rbacv1.RbacV1Client
rbacV1beta1 *rbacv1beta1.RbacV1beta1Client
rbacV1alpha1 *rbacv1alpha1.RbacV1alpha1Client
schedulingV1alpha1 *schedulingv1alpha1.SchedulingV1alpha1Client
schedulingV1beta1 *schedulingv1beta1.SchedulingV1beta1Client
schedulingV1 *schedulingv1.SchedulingV1Client
settingsV1alpha1 *settingsv1alpha1.SettingsV1alpha1Client
storageV1beta1 *storagev1beta1.StorageV1beta1Client
storageV1 *storagev1.StorageV1Client
storageV1alpha1 *storagev1alpha1.StorageV1alpha1Client
}
以CoreV1资源为例，获取k8s clientSet 结构体中对应资源版本的restclient：
restClient:= clientSet.CoreV1().RESTClient()
//1、CoreV1Client is used to interact with features provided by the group.
// CoreV1Client结构体是clientSet的成员字段的类型之一。
type CoreV1Client struct {
restClient rest.Interface
}
// 2、CoreV1 retrieves the CoreV1Client
func (c *Clientset) CoreV1() corev1.CoreV1Interface {
return c.coreV1
//coreV1即CoreV1Client结构体，coreV1为clientSet的成员字段之一，
//coreV1为 *corev1.CoreV1Client
}
type CoreV1Interface interface {
RESTClient() rest.Interface
....
}
// 3、CoreV1Client结构体为CoreV1Interface实现了RESTClient（）方法。
//returns a RESTClient that is used to communicate
// with API server by this client implementation.
func (c *CoreV1Client) RESTClient() rest.Interface {
if c == nil {
return nil
}
return c.restClient
}
// 4、 rest.Interface： Interface captures the set of operations for generically interacting with Kubernetes REST apis.
type Interface interface {
GetRateLimiter() flowcontrol.RateLimiter
Verb(verb string) *Request
Post() *Request
Put() *Request
Patch(pt types.PatchType) *Request
Get() *Request
Delete() *Request
APIVersion() schema.GroupVersion
}
//5、Most consumers should use client.New() to get a Kubernetes API client.
// rest.RESTClient结构体实现了rest.Interface接口。
type RESTClient struct {
// base is the root URL for all invocations of the client
base *url.URL
// versionedAPIPath is a path segment connecting the base URL to the resource root
versionedAPIPath string
// content describes how a RESTClient encodes and decodes responses.
content ClientContentConfig
// creates BackoffManager that is passed to requests.
createBackoffMgr func() BackoffManager
// rateLimiter is shared among all requests created by this client unless specifically
// overridden.
rateLimiter flowcontrol.RateLimiter
// Set specific behavior of the client. If not set http.DefaultClient will be used.
Client *http.Client
}
// 6、CoreV1中如何创建RESTClient结构体NewForConfig creates a new CoreV1Client for the given config.
func NewForConfig(c *rest.Config) (*CoreV1Client, error) {
config := *c
if err := setConfigDefaults(&config); err != nil {
return nil, err
}
client, err := rest.RESTClientFor(&config)
if err != nil {
return nil, err
}
return &CoreV1Client{client}, nil
}
//
client-go中client如何创建的资源？- restful
k8s的client-go提供了对资源的增删改查的接口, 我们可以调用相关资源结构体的api，进行增删改查的操作。如下以deployment示例：
1、用户获取k8s client的 deployment的调用接口
func k8sDeployment() appv1.DeploymentInterface {
return k8sClientset().AppsV1().Deployments(k8sNamespace)
}
2、相应的资源包已经实现了获取k8s相关资源的调用接口
// deployments implements DeploymentInterface
type deployments struct {
client rest.Interface
ns string
}
// newDeployments returns a Deployments
func newDeployments(c *AppsV1Client, namespace string) *deployments {
return &deployments{
client: c.RESTClient(),
ns: namespace,
}
}
//AppsV1包实现了获取deployment 资源的接口
func (c *AppsV1Client) Deployments(namespace string) DeploymentInterface {
return newDeployments(c, namespace)
}
3、client-go 中 deployments结构体已经实现的调用方法如下
// DeploymentsGetter has a method to return a DeploymentInterface.
// A group's client should implement this interface.
type DeploymentsGetter interface {
Deployments(namespace string) DeploymentInterface
}
// DeploymentInterface has methods to work with Deployment resources.
type DeploymentInterface interface {
Create(*v1.Deployment) (*v1.Deployment, error)
Update(*v1.Deployment) (*v1.Deployment, error)
UpdateStatus(*v1.Deployment) (*v1.Deployment, error)
Delete(name string, options *metav1.DeleteOptions) error
DeleteCollection(options *metav1.DeleteOptions, listOptions metav1.ListOptions) error
Get(name string, options metav1.GetOptions) (*v1.Deployment, error)
List(opts metav1.ListOptions) (*v1.DeploymentList, error)
Watch(opts metav1.ListOptions) (watch.Interface, error)
Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1.Deployment, err error)
GetScale(deploymentName string, options metav1.GetOptions) (*autoscalingv1.Scale, error)
UpdateScale(deploymentName string, scale *autoscalingv1.Scale) (*autoscalingv1.Scale, error)
DeploymentExpansion
}
4、client-go接口的底层实现是给k8s集群的apiserver发送http-restful请求。post方法的body的入参是资源obj。
}
// Create takes the representation of a deployment and creates it.
// Returns the server's representation of the deployment, and an error, if there is any.
func (c *deployments) Create(deployment *v1.Deployment) (result *v1.Deployment, err error) {
result = &v1.Deployment{}
err = c.client.Post().
Namespace(c.ns). //资源创建的ns
Resource("deployments"). //资源创建的kind
Body(deployment). //资源创建的spec
Do().
Into(result)
return
}
5、对资源的不同请求restful的method也不相同 , 常见的请求如下， 参考链接
（1）create a Deployment - HTTP Request: 用户要确定创建资源的ns和kind，在url最后两个字段体现。
POST /apis/apps/v1/namespaces/{namespace}/deployments
（2）partially update the specified Deployment - HTTP Request (部分更新)
PATCH /apis/apps/v1/namespaces/{namespace}/deployments/{name}.
（3）replace the specified Deployment - HTTP Request）（完整替换当前资源，yaml的规格要求完备）
PUT /apis/apps/v1/namespaces/{namespace}/deployments/{name}
（4）delete a Deployment - HTTP Request (用户要给出资源ns\kind\name)
DELETE /apis/apps/v1/namespaces/{namespace}/deployments/{name}
（5）read the specified Deployment - HTTP Request (用户要给出资源ns\kind\name)
GET /apis/apps/v1/namespaces/{namespace}/deployments/{name}
（6）list or watch objects of kind Deployment -HTTP Request (用户要给出资源ns\kind)
GET /apis/apps/v1/namespaces/{namespace}/deployments
k8s client-go中用到的interface和struct
1、“k8s.io/apimachinery/pkg/runtime/schema” 包中k8s schema.GroupVersionKind结构体存储group、version、kind字段的值。
// GroupVersionKind unambiguously identifies a kind. It doesn't anonymously include GroupVersion
// to avoid automatic coercion. It doesn't use a GroupVersion to avoid custom marshalling
type GroupVersionKind struct {
Group string
Version string
Kind string
}
2、"k8s.io/apimachinery/pkg/runtime"包中runtime.Object接口，k8s的资源类型都应该实现了该接口。在Scheme上注册的所有API类型都必须支持object对象接口。 由于预计方案中的对象将被序列化到wire，因此对象必须提供接口允许序列化程序设置对象表示的种类，版本和组。 在不希望序列化对象的情况下，对象可以选择返回无操作ObjectKindAccessor。
// Object interface must be supported by all API types registered with Scheme. Since objects in a scheme are
// expected to be serialized to the wire, the interface an Object must provide to the Scheme allows
// serializers to set the kind, version, and group the object is represented as. An Object may choose
// to return a no-op ObjectKindAccessor in cases where it is not expected to be serialized.
type Object interface {
GetObjectKind() schema.ObjectKind
DeepCopyObject() Object
}
3、从Scheme序列化的所有对象都对其类型信息进行编码。 序列化使用此接口，将Scheme中的类型信息设置到对象的序列化版本上。 对于无法序列化或具有独特要求的对象，此接口可能是无操作的。
// All objects that are serialized from a Scheme encode their type information. This interface is used
// by serialization to set type information from the Scheme onto the serialized version of an object.
// For objects that cannot be serialized or have unique requirements, this interface may be a no-op.
type ObjectKind interface {
// SetGroupVersionKind sets or clears the intended serialized kind of an object. Passing kind nil
// should clear the current setting.
SetGroupVersionKind(kind GroupVersionKind)
// GroupVersionKind returns the stored group, version, and kind of an object, or nil if the object does
// not expose or provide these fields.
GroupVersionKind() GroupVersionKind
}
4、runtime.RawExtension结构体，存储资源object的结构体和object的json字节流数据。
type RawExtension struct {
// Raw is the underlying serialization of this object.
//
// TODO: Determine how to detect ContentType and ContentEncoding of 'Raw' data.
Raw []byte `json:"-" protobuf:"bytes,1,opt,name=raw"`
// Object can hold a representation of this extension - useful for working with versioned
// structs.
Object Object `json:"-"`
}
+k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object 指请在生成DeepCopy时，实现Kubernetes提供的runtime.Object接口
5、“k8s.io/apimachinery/pkg/apis/meta/v1/unstructured”。 unstructuredJSONScheme 是k8s-client中重要的结构体。unstructuredJSONScheme结构体的decode方法反序列化object的json字节流数据为相应的runtime.Object，并返回相应的GroupVersionKind结构体。
// Codec is a Serializer that deals with the details of versioning objects. It offers the same
// interface as Serializer, so this is a marker to consumers that care about the version of the objects
// they receive.
type Codec Serializer
// UnstructuredJSONScheme is capable of converting JSON data into the Unstructured
// type, which can be used for generic access to objects without a predefined scheme.
// TODO: move into serializer/json.
var UnstructuredJSONScheme runtime.Codec = unstructuredJSONScheme{}
type unstructuredJSONScheme struct{}
func (s unstructuredJSONScheme) Decode(data []byte, _ *schema.GroupVersionKind, obj runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
var err error
if obj != nil {
err = s.decodeInto(data, obj)
} else {
obj, err = s.decode(data)
}
if err != nil {
return nil, nil, err
}
gvk := obj.GetObjectKind().GroupVersionKind()
if len(gvk.Kind) == 0 {
return nil, &gvk, runtime.NewMissingKindErr(string(data))
}
return obj, &gvk, nil
}
6、“k8s.io/apimachinery/pkg/util/yaml”, k8syaml文件解析包，一般使用decoder := yaml.NewYAMLOrJSONDecoder(r, 4096)，来解析原始yaml字节流请求为json字节流的形式。
// NewYAMLToJSONDecoder decodes YAML documents from the provided
// stream in chunks by converting each document (as defined by
// the YAML spec) into its own chunk, converting it to JSON via
// yaml.YAMLToJSON, and then passing it to json.Decoder.
func NewYAMLToJSONDecoder(r io.Reader) *YAMLToJSONDecoder {
reader := bufio.NewReader(r)
return &YAMLToJSONDecoder{
reader: NewYAMLReader(reader),
}
}
// Decode reads a YAML document as JSON from the stream or returns
// an error. The decoding rules match json.Unmarshal, not
// yaml.Unmarshal.
func (d *YAMLToJSONDecoder) Decode(into interface{}) error {
bytes, err := d.reader.Read()
if err != nil && err != io.EOF {
return err
}
if len(bytes) != 0 {
err := yaml.Unmarshal(bytes, into)
if err != nil {
return YAMLSyntaxError{err}
}
}
return err
}
如何解析yaml创建资源
1、把传入的yaml字节流请求进行解析，一个yaml字节流可能包含多个需要创建更新的资源object, 把yaml字节流中每个资源放到相应runtime.RawExtension{}中进行存储。
func (h *handler) decode(r io.Reader) error {
decoder := yaml.NewYAMLOrJSONDecoder(r, 4096)
for {
//提供object序列化后的存储
ext := runtime.RawExtension{}
//每次序列化一个object到ext中，把接收的yaml文件流转为json流到ext中
if err := decoder.Decode(&ext); err != nil {
if err == io.EOF {
return nil
}
return fmt.Errorf("error parsing: %v", err)
}
ext.Raw = bytes.TrimSpace(ext.Raw)
if len(ext.Raw) == 0 || bytes.Equal(ext.Raw, []byte("null")) {
continue
}
//关键步骤反序列化json数据为相应的runtime.Object，并返回相应的gkv结构
obj, gkv, err := unstructured.UnstructuredJSONScheme.Decode(ext.Raw, nil, nil)
if err != nil {
return err
}
ext.Object = obj
h.groupVersionKinds = append(h.groupVersionKinds, gkv)
h.exts = append(h.exts, &ext)
}
}
2、调用rest-client创建解析后的yaml资源
func (h *handler) apply(ctx context.Context) *metav1.Status {
type applyObject struct {
obj runtime.Object
isCreateRequest bool
restClient clientrest.Interface
namespace string
kind string
name string
}
applyObjects := make([]*applyObject, len(h.exts))
for idx, ext := range h.exts {
obj := ext.Object
gvk := h.groupVersionKinds[idx]
restClient := util.RESTClientFor(h.client, gvk.Group, gvk.Version)
namespace, err := h.metaAccessor.Namespace(obj)
if err != nil {
return errorInternal
}
name, err := h.metaAccessor.Name(obj)
if err != nil {
return errorInternal
}
genName, err := h.metaAccessor.GenerateName(obj)
if err != nil {
return errorInternal
}
if len(name) == 0 && len(genName) == 0 {
return errorBadName
}
if len(name) != 0 {
result := restClient.Get().
Context(ctx).
NamespaceIfScoped(parseNamespaceIfScoped(namespace, gvk.Kind)).
Resource(util.ResourceFromKind(gvk.Kind)).
Name(name).
Do()
err := result.Error()
if err != nil && !errors.IsNotFound(err) {
if statusError, ok := err.(*errors.StatusError); ok {
status := statusError.Status()
return &status
}
return unknownError(err)
}
if err == nil {
if h.notUpdate {
return &metav1.Status{
TypeMeta: metav1.TypeMeta{
Kind: "Status",
APIVersion: "v1",
},
Status: metav1.StatusFailure,
Code: http.StatusConflict,
Reason: metav1.StatusReasonAlreadyExists,
Details: &metav1.StatusDetails{
Name: name,
Group: gvk.Group,
Kind: gvk.Kind,
},
Message: fmt.Sprintf("%s \"%s\" already exists", gvk.Kind, name),
}
}
returnedObj, err := result.Get()
if err != nil {
return errorInternal
}
resourceVersion, err := h.metaAccessor.ResourceVersion(obj)
if err != nil {
return errorInternal
}
if resourceVersion != "" {
return errorHasResourceVersion
}
savedResourceVersion, err := h.metaAccessor.ResourceVersion(returnedObj)
if err != nil {
return errorInternal
}
if err := h.metaAccessor.SetResourceVersion(obj, savedResourceVersion); err != nil {
return errorInternal
}
applyObjects[idx] = &applyObject{
obj: obj,
isCreateRequest: false,
restClient: restClient,
namespace: namespace,
kind: gvk.Kind,
name: name,
}
continue
}
}
// create
applyObjects[idx] = &applyObject{
obj: obj,
isCreateRequest: true,
restClient: restClient,
namespace: namespace,
kind: gvk.Kind,
name: name,
}
}
var messages []string
for _, applyObj := range applyObjects {
if applyObj.isCreateRequest {
// create
result := applyObj.restClient.Post().
Context(ctx).
NamespaceIfScoped(parseNamespaceIfScoped(applyObj.namespace, applyObj.kind)).
Resource(util.ResourceFromKind(applyObj.kind)).
Body(applyObj.obj).
Do()
log.Debugf("Apply cluster bucket create call: %v", applyObj)
err := result.Error()
if err != nil {
if statusError, ok := err.(*errors.StatusError); ok {
status := statusError.Status()
return &status
}
return unknownError(err)
}
if len(applyObj.name) != 0 {
messages = append(messages, fmt.Sprintf("%s %s created", applyObj.kind, applyObj.name))
} else {
messages = append(messages, fmt.Sprintf("%s generated", applyObj.kind))
}
} else {
// update
result := applyObj.restClient.Put().
Context(ctx).
NamespaceIfScoped(parseNamespaceIfScoped(applyObj.namespace, applyObj.kind)).
Resource(util.ResourceFromKind(applyObj.kind)).
Name(applyObj.name).
Body(applyObj.obj).
Do()
log.Debugf("Apply cluster bucket update call: %v", applyObj)
err := result.Error()
if err != nil {
if statusError, ok := err.(*errors.StatusError); ok {
status := statusError.Status()
return &status
}
return unknownError(err)
}
messages = append(messages, fmt.Sprintf("%s %s configured", applyObj.kind, applyObj.name))
}
}
return &metav1.Status{
Status: metav1.StatusSuccess,
Code: http.StatusOK,
Message: strings.Join(messages, "\n"),
}
}
