一个controller怎么调用另个controller 带参数_kubernetes kube-controller-manager 源码分析...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
cmd/controller-manager/
├── OWNERS
└── app
    ├── BUILD
    ├── helper.go
    ├── helper_test.go
    ├── options
    │   ├── BUILD
    │   ├── cloudprovider.go
    │   ├── debugging.go
    │   ├── generic.go
    │   ├── kubecloudshared.go
    │   └── servicecontroller.go
    └── serve.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
cmd/kube-controller-manager/
├── BUILD
├── OWNERS
├── app
│   ├── BUILD
│   ├── apps.go
│   ├── autoscaling.go
│   ├── batch.go
│   ├── bootstrap.go
│   ├── certificates.go
│   ├── cloudproviders.go
│   ├── config
│   │   ├── BUILD
│   │   └── config.go
│   ├── controllermanager.go
│   ├── core.go
│   ├── core_test.go
│   ├── import_known_versions.go
│   ├── options
│   │   ├── BUILD
│   │   ├── attachdetachcontroller.go
│   │   ├── csrsigningcontroller.go
│   │   ├── daemonsetcontroller.go
│   │   ├── deploymentcontroller.go
│   │   ├── deprecatedcontroller.go
│   │   ├── endpointcontroller.go
│   │   ├── garbagecollectorcontroller.go
│   │   ├── hpacontroller.go
│   │   ├── jobcontroller.go
│   │   ├── namespacecontroller.go
│   │   ├── nodeipamcontroller.go
│   │   ├── nodelifecyclecontroller.go
│   │   ├── options.go
│   │   ├── options_test.go
│   │   ├── persistentvolumebindercontroller.go
│   │   ├── podgccontroller.go
│   │   ├── replicasetcontroller.go
│   │   ├── replicationcontroller.go
│   │   ├── resourcequotacontroller.go
│   │   ├── serviceaccountcontroller.go
│   │   └── ttlafterfinishedcontroller.go
│   ├── plugins.go
│   ├── policy.go
│   ├── rbac.go
│   └── testing
│       ├── BUILD
│       └── testserver.go
└── controller-manager.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
wanlei$ tree pkg/controller/ -L 1
pkg/controller/
├── BUILD
├── OWNERS
├── apis
├── bootstrap
├── certificates
├── client_builder.go
├── client_builder_dynamic.go
├── cloud
├── clusterroleaggregation
├── controller_ref_manager.go
├── controller_ref_manager_test.go
├── controller_utils.go
├── controller_utils_test.go
├── cronjob
├── daemon
├── deployment
├── disruption
├── doc.go
├── endpoint
├── garbagecollector
├── history
├── informer_factory.go
├── job
├── lookup_cache.go
├── namespace
├── nodeipam
├── nodelifecycle
├── podautoscaler
├── podgc
├── replicaset
├── replication
├── resourcequota
├── route
├── service
├── serviceaccount
├── statefulset
├── testutil
├── ttl
├── ttlafterfinished
├── util
└── volume

1
2
3
4
5
6
7
8
9
10
func main() {
    ...
    command := app.NewControllerManagerCommand()
    ...
    if err := command.Execute(); err != nil {
        fmt.Fprintf(os.Stderr, "%vn", err)
        os.Exit(1)
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// NewControllerManagerCommand creates a *cobra.Command object with default parameters
func NewControllerManagerCommand() *cobra.Command {
    s, err := options.NewKubeControllerManagerOptions()
    ...
    cmd := &cobra.Command{
        Use: "kube-controller-manager",
        Long: `The Kubernetes controller manager ....`,
        Run: func(cmd *cobra.Command, args []string) {
            ...
            c, err := s.Config(KnownControllers(), ControllersDisabledByDefault.List())
            if err != nil {
                fmt.Fprintf(os.Stderr, "%vn", err)
                os.Exit(1)
            }

            if err := Run(c.Complete(), wait.NeverStop); err != nil {
                fmt.Fprintf(os.Stderr, "%vn", err)
                os.Exit(1)
            }
        },
    }
    ...
    namedFlagSets := s.Flags(KnownControllers(), ControllersDisabledByDefault.List())
    ...
    return cmd
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
func (c *Command) Execute() error {
    _, err := c.ExecuteC()
    return err
}
func (c *Command) ExecuteC() (cmd *Command, err error) {
    // Regardless of what command execute is called on, run on Root only
    if c.HasParent() {
        return c.Root().ExecuteC()
    }
    ...
    if c.TraverseChildren {
        cmd, flags, err = c.Traverse(args)
    } else {
        cmd, flags, err = c.Find(args)
    }
    ...
    err = cmd.execute(flags)
    ...
    return cmd, err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
func (c *Command) execute(a []string) (err error) {
    ...
    for p := c; p != nil; p = p.Parent() {
        if p.PersistentPreRunE != nil {
            ...
        } else if p.PersistentPreRun != nil {
            ...
        }
    }
    if c.PreRunE != nil {
        ...
    } else if c.PreRun != nil {
        ...
    }
    ..
    if c.RunE != nil {
        ...
    } else {
        c.Run(c, argWoFlags)
    }
    if c.PostRunE != nil {
        ...
    } else if c.PostRun != nil {
        ...
    }
    for p := c; p != nil; p = p.Parent() {
        if p.PersistentPostRunE != nil {
            ...
        } else if p.PersistentPostRun != nil {
            ...
        }
    }
    return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
// Run runs the KubeControllerManagerOptions.  This should never exit.
func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
    ....
    // Setup any healthz checks we will want to use.
    var checks []healthz.HealthzChecker
    var electionChecker *leaderelection.HealthzAdaptor
    if c.ComponentConfig.Generic.LeaderElection.LeaderElect {
        electionChecker = leaderelection.NewLeaderHealthzAdaptor(time.Second * 20)
        checks = append(checks, electionChecker)
    }

    // Start the controller manager HTTP server
    // unsecuredMux is the handler for these controller *after* authn/authz filters have been applied
    var unsecuredMux *mux.PathRecorderMux
    if c.SecureServing != nil {
        unsecuredMux = genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
        handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, &c.Authorization, &c.Authentication)
        // TODO: handle stoppedCh returned by c.SecureServing.Serve
        if _, err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
            return err
        }
    }
    if c.InsecureServing != nil {
        unsecuredMux = genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
        insecureSuperuserAuthn := server.AuthenticationInfo{Authenticator: &server.InsecureSuperuser{}}
        handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, nil, &insecureSuperuserAuthn)
        if err := c.InsecureServing.Serve(handler, 0, stopCh); err != nil {
            return err
        }
    }

    run := func(ctx context.Context) {
        ...
    }

    if !c.ComponentConfig.Generic.LeaderElection.LeaderElect {
        run(context.TODO())
        panic("unreachable")
    }
    ....
    leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
        Lock:          rl,
        LeaseDuration: c.ComponentConfig.Generic.LeaderElection.LeaseDuration.Duration,
        RenewDeadline: c.ComponentConfig.Generic.LeaderElection.RenewDeadline.Duration,
        RetryPeriod:   c.ComponentConfig.Generic.LeaderElection.RetryPeriod.Duration,
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: run,
            OnStoppedLeading: func() {
                klog.Fatalf("leaderelection lost")
            },
        },
        WatchDog: electionChecker,
        Name:     "kube-controller-manager",
    })
    panic("unreachable")
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
run := func(ctx context.Context) {
        ....
        controllerContext, err := CreateControllerContext(c, rootClientBuilder, clientBuilder, ctx.Done())
        ...
        if err := StartControllers(controllerContext, saTokenControllerInitFunc, NewControllerInitializers(controllerContext.LoopMode), unsecuredMux); err != nil {
            klog.Fatalf("error starting controllers: %v", err)
        }

        controllerContext.InformerFactory.Start(controllerContext.Stop)
        controllerContext.GenericInformerFactory.Start(controllerContext.Stop)
        close(controllerContext.InformersStarted)

        select {}
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
// StartControllers starts a set of controllers with a specified ControllerContext
func StartControllers(ctx ControllerContext, startSATokenController InitFunc, controllers map[string]InitFunc, unsecuredMux *mux.PathRecorderMux) error {
    ...
    for controllerName, initFn := range controllers {
        if !ctx.IsControllerEnabled(controllerName) {
            klog.Warningf("%q is disabled", controllerName)
            continue
        }

        time.Sleep(wait.Jitter(ctx.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))

        klog.V(1).Infof("Starting %q", controllerName)
        debugHandler, started, err := initFn(ctx)
        if err != nil {
            klog.Errorf("Error starting %q", controllerName)
            return err
        }
        if !started {
            klog.Warningf("Skipping %q", controllerName)
            continue
        }
        if debugHandler != nil && unsecuredMux != nil {
            basePath := "/debug/controllers/" + controllerName
            unsecuredMux.UnlistedHandle(basePath, http.StripPrefix(basePath, debugHandler))
            unsecuredMux.UnlistedHandlePrefix(basePath+"/", http.StripPrefix(basePath, debugHandler))
        }
        klog.Infof("Started %q", controllerName)
    }

    return nil
}

// NewControllerInitializers is a public map of named controller groups (you can start more than one in an init func)
// paired to their InitFunc.  This allows for structured downstream composition and subdivision.
func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    controllers := map[string]InitFunc{}
    controllers["endpoint"] = startEndpointController
    controllers["replicationcontroller"] = startReplicationController
    controllers["podgc"] = startPodGCController
    controllers["resourcequota"] = startResourceQuotaController
    controllers["namespace"] = startNamespaceController
    controllers["serviceaccount"] = startServiceAccountController
    controllers["garbagecollector"] = startGarbageCollectorController
    controllers["daemonset"] = startDaemonSetController
    controllers["job"] = startJobController
    controllers["deployment"] = startDeploymentController
    controllers["replicaset"] = startReplicaSetController
    controllers["horizontalpodautoscaling"] = startHPAController
    controllers["disruption"] = startDisruptionController
    controllers["statefulset"] = startStatefulSetController
    controllers["cronjob"] = startCronJobController
    controllers["csrsigning"] = startCSRSigningController
    controllers["csrapproving"] = startCSRApprovingController
    controllers["csrcleaner"] = startCSRCleanerController
    controllers["ttl"] = startTTLController
    controllers["bootstrapsigner"] = startBootstrapSignerController
    controllers["tokencleaner"] = startTokenCleanerController
    controllers["nodeipam"] = startNodeIpamController
    controllers["nodelifecycle"] = startNodeLifecycleController
    if loopMode == IncludeCloudLoops {
        controllers["service"] = startServiceController
        controllers["route"] = startRouteController
        controllers["cloud-node-lifecycle"] = startCloudNodeLifecycleController
        // TODO: volume controller into the IncludeCloudLoops only set.
    }
    controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
    controllers["attachdetach"] = startAttachDetachController
    controllers["persistentvolume-expander"] = startVolumeExpandController
    controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
    controllers["pvc-protection"] = startPVCProtectionController
    controllers["pv-protection"] = startPVProtectionController
    controllers["ttl-after-finished"] = startTTLAfterFinishedController
    controllers["root-ca-cert-publisher"] = startRootCACertPublisher

    return controllers
}

1
2
3
4
5
6
7
8
9
10
11
12
func startJobController(ctx ControllerContext) (http.Handler, bool, error) {
    if !ctx.AvailableResources[schema.GroupVersionResource{Group: "batch", Version: "v1", Resource: "jobs"}] {
        return nil, false, nil
    }
    go job.NewJobController(
        ctx.InformerFactory.Core().V1().Pods(),
        ctx.InformerFactory.Batch().V1().Jobs(),
        ctx.ClientBuilder.ClientOrDie("job-controller"),
    ).Run(int(ctx.ComponentConfig.JobController.ConcurrentJobSyncs), ctx.Stop)
    return nil, true, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// StartControllers starts a set of controllers with a specified ControllerContext
func StartControllers(ctx ControllerContext, startSATokenController InitFunc, controllers map[string]InitFunc, unsecuredMux *mux.PathRecorderMux) error {
    ...
    for controllerName, initFn := range controllers {
        ...
        debugHandler, started, err := initFn(ctx)
        if err != nil {
            klog.Errorf("Error starting %q", controllerName)
            return err
        }
        if !started {
            klog.Warningf("Skipping %q", controllerName)
            continue
        }
        if debugHandler != nil && unsecuredMux != nil {
            basePath := "/debug/controllers/" + controllerName
            unsecuredMux.UnlistedHandle(basePath, http.StripPrefix(basePath, debugHandler))
            unsecuredMux.UnlistedHandlePrefix(basePath+"/", http.StripPrefix(basePath, debugHandler))
        }
        klog.Infof("Started %q", controllerName)
    }

    return nil
}

func startJobController(ctx ControllerContext) (http.Handler, bool, error) {
    ...
}