kubectl源码分析之wait

淡淡的烟草味﹌ 2023-02-20 04:46 5阅读 0赞

#  欢迎关注我的公众号： #

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAaHhwamF2YTE_size_8_color_FFFFFF_t_70_g_se_x_16][]

目前刚开始写一个月，一共写了18篇原创文章，文章目录如下：

[istio多集群探秘，部署了50次多集群后我得出的结论][istio_50]

[istio多集群链路追踪，附实操视频][istio]

[istio防故障利器，你知道几个,istio新手不要读，太难！][istio_istio]

[istio业务权限控制，原来可以这么玩][istio 1]

[istio实现非侵入压缩，微服务之间如何实现压缩][istio 2]

[不懂envoyfilter也敢说精通istio系列-http-rbac-不要只会用AuthorizationPolicy配置权限][envoyfilter_istio_-http-rbac-_AuthorizationPolicy]

[不懂envoyfilter也敢说精通istio系列-02-http-corsFilter-不要只会vs][envoyfilter_istio_-02-http-corsFilter-_vs]

[不懂envoyfilter也敢说精通istio系列-03-http-csrf filter-再也不用再代码里写csrf逻辑了][envoyfilter_istio_-03-http-csrf filter-_csrf]

[不懂envoyfilter也敢说精通istio系列http-jwt\_authn-不要只会RequestAuthorization][envoyfilter_istio_http-jwt_authn-_RequestAuthorization]

[不懂envoyfilter也敢说精通istio系列-05-fault-filter-故障注入不止是vs][envoyfilter_istio_-05-fault-filter-_vs]

[不懂envoyfilter也敢说精通istio系列-06-http-match-配置路由不只是vs][envoyfilter_istio_-06-http-match-_vs]

[不懂envoyfilter也敢说精通istio系列-07-负载均衡配置不止是dr][envoyfilter_istio_-07-_dr]

[不懂envoyfilter也敢说精通istio系列-08-连接池和断路器][envoyfilter_istio_-08-]

[不懂envoyfilter也敢说精通istio系列-09-http-route filter][envoyfilter_istio_-09-http-route filter]

[不懂envoyfilter也敢说精通istio系列-network filter-redis proxy][envoyfilter_istio_-network filter-redis proxy]

[不懂envoyfilter也敢说精通istio系列-network filter-HttpConnectionManager][envoyfilter_istio_-network filter-HttpConnectionManager]

[不懂envoyfilter也敢说精通istio系列-ratelimit-istio ratelimit完全手册][envoyfilter_istio_-ratelimit-istio ratelimit]

————————————————

type WaitFlags struct {//wait flags
    	RESTClientGetter     genericclioptions.RESTClientGetter
    	PrintFlags           *genericclioptions.PrintFlags
    	ResourceBuilderFlags *genericclioptions.ResourceBuilderFlags
    
    	Timeout      time.Duration
    	ForCondition string
    
    	genericclioptions.IOStreams
    }
    
    // NewWaitFlags returns a default WaitFlags
    func NewWaitFlags(restClientGetter genericclioptions.RESTClientGetter, streams genericclioptions.IOStreams) *WaitFlags {
    	return &WaitFlags{//初始化wait flag
    		RESTClientGetter: restClientGetter,
    		PrintFlags:       genericclioptions.NewPrintFlags("condition met"),
    		ResourceBuilderFlags: genericclioptions.NewResourceBuilderFlags().
    			WithLabelSelector("").
    			WithFieldSelector("").
    			WithAll(false).
    			WithAllNamespaces(false).
    			WithLocal(false).
    			WithLatest(),
    
    		Timeout: 30 * time.Second,
    
    		IOStreams: streams,
    	}
    }

//创建wait命令
    func NewCmdWait(restClientGetter genericclioptions.RESTClientGetter, streams genericclioptions.IOStreams) *cobra.Command {
    	flags := NewWaitFlags(restClientGetter, streams)//初始化wait flag
    
    	cmd := &cobra.Command{//创建cobra命令
    		Use:     "wait ([-f FILENAME] | resource.group/resource.name | resource.group [(-l label | --all)]) [--for=delete|--for condition=available]",
    		Short:   "Experimental: Wait for a specific condition on one or many resources.",
    		Long:    waitLong,
    		Example: waitExample,
    
    		DisableFlagsInUseLine: true,
    		Run: func(cmd *cobra.Command, args []string) {
    			o, err := flags.ToOptions(args)//wait flag转waitOption
    			cmdutil.CheckErr(err)
    			err = o.RunWait()//运行
    			cmdutil.CheckErr(err)
    		},
    		SuggestFor: []string{"list", "ps"},
    	}
    
    	flags.AddFlags(cmd)//添加选项
    
    	return cmd
    }
    
    // AddFlags registers flags for a cli
    func (flags *WaitFlags) AddFlags(cmd *cobra.Command) {//选项
    	flags.PrintFlags.AddFlags(cmd)//打印选项
    	flags.ResourceBuilderFlags.AddFlags(cmd.Flags())//resource builder选项
    
    	cmd.Flags().DurationVar(&flags.Timeout, "timeout", flags.Timeout, "The length of time to wait before giving up.  Zero means check once and don't wait, negative means wait for a week.")//timeout选项
    	cmd.Flags().StringVar(&flags.ForCondition, "for", flags.ForCondition, "The condition to wait on: [delete|condition=condition-name].")//for选项
    }
    
    // ToOptions converts from CLI inputs to runtime inputs
    func (flags *WaitFlags) ToOptions(args []string) (*WaitOptions, error) {//wait flag转option
    	printer, err := flags.PrintFlags.ToPrinter()//print flag转printer
    	if err != nil {
    		return nil, err
    	}
    	builder := flags.ResourceBuilderFlags.ToBuilder(flags.RESTClientGetter, args)//创建builder
    	clientConfig, err := flags.RESTClientGetter.ToRESTConfig()//获取restconfig
    	if err != nil {
    		return nil, err
    	}
    	dynamicClient, err := dynamic.NewForConfig(clientConfig)//获取dynamicClient
    	if err != nil {
    		return nil, err
    	}
    	conditionFn, err := conditionFuncFor(flags.ForCondition, flags.ErrOut)//获取condition函数
    	if err != nil {
    		return nil, err
    	}
    
    	effectiveTimeout := flags.Timeout
    	if effectiveTimeout < 0 {//设置超时时间
    		effectiveTimeout = 168 * time.Hour
    	}
    
    	o := &WaitOptions{//构造wait option
    		ResourceFinder: builder,
    		DynamicClient:  dynamicClient,
    		Timeout:        effectiveTimeout,
    
    		Printer:     printer,
    		ConditionFn: conditionFn,
    		IOStreams:   flags.IOStreams,
    	}
    
    	return o, nil
    }

//获取condition函数
    func conditionFuncFor(condition string, errOut io.Writer) (ConditionFunc, error) {
    	if strings.ToLower(condition) == "delete" {//如果条件是delete，则返回IsDeleted函数
    		return IsDeleted, nil
    	}
    	if strings.HasPrefix(condition, "condition=") {//如果有condition=前缀
    		conditionName := condition[len("condition="):]//获取condition名称
    		conditionValue := "true"//设置conditionvalue
    		if equalsIndex := strings.Index(conditionName, "="); equalsIndex != -1 {
    			conditionValue = conditionName[equalsIndex+1:]
    			conditionName = conditionName[0:equalsIndex]
    		}
    
    		return ConditionalWait{//构造conditionWait，返回isConditionMet函数
    			conditionName:   conditionName,
    			conditionStatus: conditionValue,
    			errOut:          errOut,
    		}.IsConditionMet, nil
    	}
    
    	return nil, fmt.Errorf("unrecognized condition: %q", condition)
    }

//运行
    func (o *WaitOptions) RunWait() error {
    	visitCount := 0
    	err := o.ResourceFinder.Do().Visit(func(info *resource.Info, err error) error {// visit result
    		if err != nil {
    			return err
    		}
    
    		visitCount++
    		finalObject, success, err := o.ConditionFn(info, o)//返回condition状态
    		if success {//如果是success，打印对象
    			o.Printer.PrintObj(finalObject, o.Out)
    			return nil
    		}
    		if err == nil {
    			return fmt.Errorf("%v unsatisified for unknown reason", finalObject)
    		}
    		return err
    	})
    	if err != nil {
    		return err
    	}
    	if visitCount == 0 {
    		return errNoMatchingResources
    	}
    	return err
    }

//删除condition函数
    func IsDeleted(info *resource.Info, o *WaitOptions) (runtime.Object, bool, error) {
    	endTime := time.Now().Add(o.Timeout)//设置结束时间
    	for {
    		if len(info.Name) == 0 {//名称为空返回错误
    			return info.Object, false, fmt.Errorf("resource name must be provided")
    		}
    
    		nameSelector := fields.OneTermEqualSelector("metadata.name", info.Name).String()//构造field-selector
    
    		// List with a name field selector to get the current resourceVersion to watch from (not the object's resourceVersion)
    		gottenObjList, err := o.DynamicClient.Resource(info.Mapping.Resource).Namespace(info.Namespace).List(metav1.ListOptions{FieldSelector: nameSelector})//删选资源
    		if apierrors.IsNotFound(err) {//如果未找到资源返回true
    			return info.Object, true, nil
    		}
    		if err != nil {//有错误返回错误
    			// TODO this could do something slightly fancier if we wish
    			return info.Object, false, err
    		}
    		if len(gottenObjList.Items) != 1 {//items不为一个，返回true
    			return info.Object, true, nil
    		}
    		gottenObj := &gottenObjList.Items[0]
    		resourceLocation := ResourceLocation{//构造resourceLocation
    			GroupResource: info.Mapping.Resource.GroupResource(),
    			Namespace:     gottenObj.GetNamespace(),
    			Name:          gottenObj.GetName(),
    		}
    		if uid, ok := o.UIDMap[resourceLocation]; ok {
    			if gottenObj.GetUID() != uid {//uid不等，返回true
    				return gottenObj, true, nil
    			}
    		}
    
    		watchOptions := metav1.ListOptions{}//构造wait结构体
    		watchOptions.FieldSelector = nameSelector
    		watchOptions.ResourceVersion = gottenObjList.GetResourceVersion()
    		objWatch, err := o.DynamicClient.Resource(info.Mapping.Resource).Namespace(info.Namespace).Watch(watchOptions)//获取watch对象
    		if err != nil {
    			return gottenObj, false, err
    		}
    
    		timeout := endTime.Sub(time.Now())
    		errWaitTimeoutWithName := extendErrWaitTimeout(wait.ErrWaitTimeout, info)
    		if timeout < 0 {//超时，返回错误
    			// we're out of time
    			return gottenObj, false, errWaitTimeoutWithName
    		}
    
    		ctx, cancel := watchtools.ContextWithOptionalTimeout(context.Background(), o.Timeout)//获取ctx和cancel
    		watchEvent, err := watchtools.UntilWithoutRetry(ctx, objWatch, Wait{errOut: o.ErrOut}.IsDeleted)//等待条件满足
    		cancel()
    		switch {
    		case err == nil:// 错误为nil返回true
    			return watchEvent.Object, true, nil
    		case err == watchtools.ErrWatchClosed:
    			continue
    		case err == wait.ErrWaitTimeout:
    			if watchEvent != nil {
    				return watchEvent.Object, false, errWaitTimeoutWithName
    			}
    			return gottenObj, false, errWaitTimeoutWithName
    		default:
    			return gottenObj, false, err
    		}
    	}
    }

//判断是否否删除完成
    func (w Wait) IsDeleted(event watch.Event) (bool, error) {
    	switch event.Type {//判断事件类型
    	case watch.Error://有错误打印错误，返回false
    		// keep waiting in the event we see an error - we expect the watch to be closed by
    		// the server if the error is unrecoverable.
    		err := apierrors.FromObject(event.Object)
    		fmt.Fprintf(w.errOut, "error: An error occurred while waiting for the object to be deleted: %v", err)
    		return false, nil
    	case watch.Deleted://delete，返回true
    		return true, nil
    	default:
    		return false, nil//默认返回false
    	}
    }

//条件函数
    func (w ConditionalWait) IsConditionMet(info *resource.Info, o *WaitOptions) (runtime.Object, bool, error) {
    	endTime := time.Now().Add(o.Timeout)//设置超时时间
    	for {
    		if len(info.Name) == 0 {//name为空返回错误
    			return info.Object, false, fmt.Errorf("resource name must be provided")
    		}
    
    		nameSelector := fields.OneTermEqualSelector("metadata.name", info.Name).String()//构造字段选择器
    
    		var gottenObj *unstructured.Unstructured
    		// List with a name field selector to get the current resourceVersion to watch from (not the object's resourceVersion)
    		gottenObjList, err := o.DynamicClient.Resource(info.Mapping.Resource).Namespace(info.Namespace).List(metav1.ListOptions{FieldSelector: nameSelector})//删选资源
    
    		resourceVersion := ""
    		switch {
    		case err != nil://err不为空返回错误
    			return info.Object, false, err
    		case len(gottenObjList.Items) != 1://items不为1个
    			resourceVersion = gottenObjList.GetResourceVersion()
    		default:
    			gottenObj = &gottenObjList.Items[0]//获取obj
    			conditionMet, err := w.checkCondition(gottenObj)//判断条件是否满足
    			if conditionMet {//如果满足返回true
    				return gottenObj, true, nil
    			}
    			if err != nil {//有错误返回错误
    				return gottenObj, false, err
    			}
    			resourceVersion = gottenObjList.GetResourceVersion()//设置resourceVersion
    		}
    
    		watchOptions := metav1.ListOptions{}//构造waitoption
    		watchOptions.FieldSelector = nameSelector
    		watchOptions.ResourceVersion = resourceVersion
    		objWatch, err := o.DynamicClient.Resource(info.Mapping.Resource).Namespace(info.Namespace).Watch(watchOptions)//获取watch对象
    		if err != nil {
    			return gottenObj, false, err
    		}
    
    		timeout := endTime.Sub(time.Now())
    		errWaitTimeoutWithName := extendErrWaitTimeout(wait.ErrWaitTimeout, info)
    		if timeout < 0 {//超时返回错误
    			// we're out of time
    			return gottenObj, false, errWaitTimeoutWithName
    		}
    
    		ctx, cancel := watchtools.ContextWithOptionalTimeout(context.Background(), o.Timeout)//获取ctx和cancle
    		watchEvent, err := watchtools.UntilWithoutRetry(ctx, objWatch, w.isConditionMet)//等待直到条件满足
    		cancel()
    		switch {
    		case err == nil://错误为nil返回true
    			return watchEvent.Object, true, nil
    		case err == watchtools.ErrWatchClosed:
    			continue
    		case err == wait.ErrWaitTimeout:
    			if watchEvent != nil {
    				return watchEvent.Object, false, errWaitTimeoutWithName
    			}
    			return gottenObj, false, errWaitTimeoutWithName
    		default:
    			return gottenObj, false, err
    		}
    	}
    }

//判断条件是否满足
    func (w ConditionalWait) checkCondition(obj *unstructured.Unstructured) (bool, error) {
    	conditions, found, err := unstructured.NestedSlice(obj.Object, "status", "conditions")//获取conditions
    	if err != nil {
    		return false, err
    	}
    	if !found {
    		return false, nil
    	}
    	for _, conditionUncast := range conditions {//遍历conditions
    		condition := conditionUncast.(map[string]interface{})//获取condition
    		name, found, err := unstructured.NestedString(condition, "type")//获取condition名称
    		if !found || err != nil || strings.ToLower(name) != strings.ToLower(w.conditionName) {//condition名称不是我们要找的名称跳过
    			continue
    		}
    		status, found, err := unstructured.NestedString(condition, "status")//获取状态
    		if !found || err != nil {
    			continue
    		}
    		return strings.ToLower(status) == strings.ToLower(w.conditionStatus), nil//判断状态是否是我们指定的状态
    	}
    
    	return false, nil
    }
    
    //判断条件是否满足
    func (w ConditionalWait) isConditionMet(event watch.Event) (bool, error) {
    	if event.Type == watch.Error {//如果事件类型是错误，打印错误，返回false
    		// keep waiting in the event we see an error - we expect the watch to be closed by
    		// the server
    		err := apierrors.FromObject(event.Object)
    		fmt.Fprintf(w.errOut, "error: An error occurred while waiting for the condition to be satisfied: %v", err)
    		return false, nil
    	}
    	if event.Type == watch.Deleted {//如果事件类型是delete，返回false
    		// this will chain back out, result in another get and an return false back up the chain
    		return false, nil
    	}
    	obj := event.Object.(*unstructured.Unstructured)
    	return w.checkCondition(obj)//检查状态是否满足
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAaHhwamF2YTE_size_8_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/20211007094240106.jpg?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAaHhwamF2YTE=,size_8,color_FFFFFF,t_70,g_se,x_16
[istio_50]: https://mp.weixin.qq.com/s/_OPMjmWCDSTlmbAxytg33A
[istio]: https://mp.weixin.qq.com/s/kjvtno_UzJAGr1joxm3tvw
[istio_istio]: https://mp.weixin.qq.com/s/e2-p_NY31Mc5y07S4Bb1JQ
[istio 1]: https://mp.weixin.qq.com/s/iTV0SCQLhSOTYncp3PxoHQ
[istio 2]: https://mp.weixin.qq.com/s/eINlrT242RlAwfuZu1_PcA
[envoyfilter_istio_-http-rbac-_AuthorizationPolicy]: https://mp.weixin.qq.com/s/Q-yTGBNqkjOov49LTXPrPA
[envoyfilter_istio_-02-http-corsFilter-_vs]: https://mp.weixin.qq.com/s/bFMdrT7OKSqoWv8pDXOIqQ
[envoyfilter_istio_-03-http-csrf filter-_csrf]: https://mp.weixin.qq.com/s/dlG9phV-kMMxs4UPzuRNYg
[envoyfilter_istio_http-jwt_authn-_RequestAuthorization]: https://mp.weixin.qq.com/s/IfrJ-FdLYsdhPkDrcO_YXA
[envoyfilter_istio_-05-fault-filter-_vs]: https://mp.weixin.qq.com/s/a4gDXqD8uswgglMwkUph2Q
[envoyfilter_istio_-06-http-match-_vs]: https://mp.weixin.qq.com/s/TpS3DHdcaJiPivIS_jsIGw
[envoyfilter_istio_-07-_dr]: https://mp.weixin.qq.com/s/k-zJm7n7OOKg6MotQJYPAA
[envoyfilter_istio_-08-]: https://mp.weixin.qq.com/s/n_bJ3saNeTtXRprIymhwzg
[envoyfilter_istio_-09-http-route filter]: https://mp.weixin.qq.com/s/JAeZr0BMVQxIrB3S_7Ru0A
[envoyfilter_istio_-network filter-redis proxy]: https://mp.weixin.qq.com/s/B9IPTI8F3q42nVkqbQRlPw
[envoyfilter_istio_-network filter-HttpConnectionManager]: https://mp.weixin.qq.com/s/W3OPQWkCVvuHGS49cTNhlg
[envoyfilter_istio_-ratelimit-istio ratelimit]: https://mp.weixin.qq.com/s/KgiDsELNol27mgZvDbFsqw