Function OperatorsCmd

operator/cmd.go:27–119

()

)

func OperatorsCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:          "operator",
		Short:        "kubernetes operators control",
		SilenceUsage: true,
		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
			zconf := zap.NewDevelopmentConfig()
			zconf.DisableCaller = true
			zconf.DisableStacktrace = true
			zconf.EncoderConfig.EncodeTime = func(time.Time, zapcore.PrimitiveArrayEncoder) {}

			zapLog, _ := zconf.Build()

			group, ctx := errgroup.WithContext(cmd.Context())

			cmd.SetContext(logr.NewContext(ctx, zapr.NewLogger(zapLog)))

			kubecfg, err := fromctx.KubeConfigFromCtx(cmd.Context())
			if err != nil {
				if err := clientcommon.SetKubeConfigToCmd(cmd); err != nil {
					return err
				}

				kubecfg = fromctx.MustKubeConfigFromCtx(cmd.Context())
			}

			if val := ctx.Value(fromctx.CtxKeyKubeClientSet); val == nil {
				kc, err := kubernetes.NewForConfig(kubecfg)
				if err != nil {
					return err
				}
				fromctx.CmdSetContextValue(cmd, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
			}

			if val := ctx.Value(fromctx.CtxKeyAkashClientSet); val == nil {
				ac, err := akashclientset.NewForConfig(kubecfg)
				if err != nil {
					return err
				}

				fromctx.CmdSetContextValue(cmd, fromctx.CtxKeyAkashClientSet, akashclientset.Interface(ac))
			}

			startupch := make(chan struct{}, 1)
			fromctx.CmdSetContextValue(cmd, fromctx.CtxKeyStartupCh, (chan<- struct{})(startupch))

			pctx, pcancel := context.WithCancel(context.Background())

			fromctx.CmdSetContextValue(cmd, fromctx.CtxKeyErrGroup, group)
			fromctx.CmdSetContextValue(cmd, fromctx.CtxKeyPubSub, pubsub.New(pctx, 1000))

			go func() {
				defer pcancel()

				select {
				case <-ctx.Done():
					return
				case <-startupch: