kube-proxyの実装から学ぶiptables 〜 Part 1. kube-proxyの基礎 〜

iptables -A INPUT -s 192.0.2.1 -j DROP
また、SSH 通信のみを許可する場合は以下の様になります。
iptables -A INPUT -p tcp -m tcp --dport 22 -j ACCEPT
また、iptables はルールによってパケットの中身を変更することもでき、例えば特定のパケットの宛先を変更する場合 (Destination-NAT) は以下の様なルールを設定します。
iptables -A INPUT -d 192.0.2.1/32 -p tcp -m tcp -j DNAT --to-destination 192.0.2.2:80
iptables プロキシモードの kube-proxy は Service リソースの状態に応じて各ワーカーノードで iptables コマンドを実行し、Service の IPアドレス宛のトラフィックをその Service に紐づく Pod にリダイレクトするといったことをしています。
kube-proxy の実装
実装の概要
ここからは kube-proxy の実装について確認していきます。kube-proxy は他の Kubernetes コンポーネントと同様に Go で実装されており、そのソースコードは https://github.com/kubernetes/kubernetes のリポジトリにあります。今回は v1.28.4 の実装を対象とします。
具体的な実装に入る前に、kube-proxy の実装の概要について確認しておきましょう。図にすると以下の様になります。
kube-proxy の実体が ProxyServer という構造体で、Client や Proxier と言った構造体を持っています (他にも色々持っていますが省略しています)。
ProxyServer は Client を使って kube-apiserver を通して Service や EndpointSlice というリソースを常時監視しています。それらのリソースが作成・変更されると ProxyServer はそれを検知し、Proxier の syncProxyRules() を実行してリソースの状態とワーカーノードの iptables のルールを同期させます。また、Proxier は起動時に syncLoop() を実行し、リソースの変更がなくても一定時間ごとに syncProxyRules() を実行し、リソースの状態と iptables のルール同期を行います。
ProxyServer が持つ Proxier は https://github.com/kubernetes/kubernetes/blob/v1.28.4/pkg/proxy/types.go#L30 で定義されている proxy.Provider インターフェースであり、 Service/EndpointSlice リソース変更をハンドリングするための config.ServiceHandler, config.EndpointSliceHandler インターフェースや 、syncProxyRules() のラッパーである sync() 、sync() を定期実行するための syncLoop() を持ちます。
types.go
// #L30
type Provider interface {
	config.EndpointSliceHandler
	config.ServiceHandler
	config.NodeHandler
	// Sync immediately synchronizes the Provider's current state to proxy rules.
	Sync()
	// SyncLoop runs periodic work.
	// This is expected to run as a goroutine or as the main loop of the app.
	// It does not return.
	SyncLoop()
kube-proxy の動作モードによって Proxier の実装は異なり、例えば今回対象としている iptables プロキシモードの場合は https://github.com/kubernetes/kubernetes/blob/v1.28.4/pkg/proxy/iptables/proxier.go が Proxier の実装となります。
エントリーポイント
それでは kube-proxy の実装をエントリーポイントから追っていきましょう。
kube-proxy のエントリーポイントは https://github.com/kubernetes/kubernetes/blob/v1.28.4/cmd/kube-proxy/proxy.go にあり、ここでは cobra.Command を作って Run() しているだけです。
proxy.go




    

func main() {
	command := app.NewProxyCommand()
	code := cli.Run(command)
	os.Exit(code)
ProxyServerの作成と起動
エントリーポイントで作成される cobra.Command には ProxyServer の作成処理が登録されており、その実装は https://github.com/kubernetes/kubernetes/blob/v1.28.4/cmd/kube-proxy/app/server.go#L358 で newProxyServer() によって ProxyServer の作成を行い、最後の o.runLoop() の実行によって作成した ProxyServer を起動しています。
server.go
// #L358
// Run runs the specified ProxyServer.
func (o *Options) Run() error {
	defer close(o.errCh)
	if len(o.WriteConfigTo) > 0 {
		return o.writeConfigFile()
	err := platformCleanup(o.config.Mode, o.CleanupAndExit)
	if o.CleanupAndExit {
		return err
	// We ignore err otherwise; the cleanup is best-effort, and the backends will have
	// logged messages if they failed in interesting ways.
	proxyServer, err := newProxyServer(o.config, o.master, o.InitAndExit)
	if err != nil {
		return err
	if o.InitAndExit {
		return nil
	o.proxyServer = proxyServer
	return o.runLoop()
server.go
// #L378
// runLoop will watch on the update change of the proxy server's configuration file.
// Return an error when updated
func (o *Options) runLoop() error {
	if o.watcher != nil {
		o.watcher.Run()
	// run the proxy in goroutine
	go func() {
		err := o.proxyServer.Run()
		o.errCh <- err
	for {
		err := <-o.errCh
		if err != nil {
			return err
newProxyServer() の中では kube-apiserver を通して Service リソース等を監視するための Client の作成や、実際にネットワーク設定を行う Proxier 構造体の作成が createClient() や createProxier() によって行われています。
server.go
// #L581
// newProxyServer creates a ProxyServer based on the given config
func newProxyServer(config *kubeproxyconfig.KubeProxyConfiguration, master string, initOnly bool) (*ProxyServer, error) {
	s := &ProxyServer{Config: config}
    // (中略)
	s.Client, err = createClient(config.ClientConnection, master)
	if err != nil {
		return nil, err
	// (中略)
	s.Proxier, err = s.createProxier(config, dualStackSupported, initOnly)
	if err != nil {
		return nil, err
	return s, nil
Proxierの作成と起動
createProxier() の実装は https://github.com/kubernetes/kubernetes/blob/v1.28.4/cmd/kube-proxy/app/server_others.go#L127 にあり、kube-proxy の動作モードや dualStack (ipv4/6 の両方を利用) かどうかによって分岐し、どの Proxier の実装を返すかを選択しています。singleStack の iptables プロキシモードの場合の実装は以下の部分になります。
server_others.go
// #L127
// createProxier creates the proxy.Provider
func (s *ProxyServer) createProxier(config *proxyconfigapi.KubeProxyConfiguration, dualStack bool) (proxy.Provider, error) {
	var proxier proxy.Provider
	var err error
	// (中略)
	if config.Mode == proxyconfigapi.ProxyModeIPTables {
		klog.InfoS("Using iptables Proxier")
		if dualStack {
			// dualStack の場合の Proxier 作成処理
            // (中略)
		} else {
			// Create a single-stack proxier if and only if the node does not support dual-stack (i.e, no iptables support).
			var localDetector proxyutiliptables.LocalTrafficDetector
			localDetector, err = getLocalDetector(s.PrimaryIPFamily, config.DetectLocalMode, config, s.podCIDRs)
			if err != nil {
				return nil, fmt.Errorf("unable to create proxier: %v", err)
			// TODO this has side effects that should only happen when Run() is invoked.
			proxier, err = iptables.NewProxier(
				s.PrimaryIPFamily,
				iptInterface,
				utilsysctl.New(),
				execer,
				config.IPTables.SyncPeriod.Duration,
				config.IPTables.MinSyncPeriod.Duration,
				config.IPTables.MasqueradeAll,
				*config.IPTables.LocalhostNodePorts,
				int(*config.IPTables.MasqueradeBit),
				localDetector,
				s.Hostname,
				s.NodeIPs[s.PrimaryIPFamily],
				s.Recorder,
				s.HealthzServer,
				config.NodePortAddresses,
		if err != nil {
			return nil, fmt.Errorf("unable to create proxier: %v", err)
	} else if config.Mode == proxyconfigapi.ProxyModeIPVS {
		// IPVA プロキシモードの場合の Proxier 作成処理
        // (中略)
	return proxier, nil
iptables.NewProxier() が iptables を利用する Proxier を作成するメソッドで、https://github.com/kubernetes/kubernetes/blob/v1.28.4/pkg/proxy/iptables/proxier.go#L222 に実装があります。Proxier の syncRunner フィールドに代入する構造体を返す async.NewBoundedFrequencyRunner() に自身の syncProxyRules() を渡すことで、sync() 実行時に synrRunner を通して syncProxyRules() が実行される様にしています。
proxier.go
// #L222
// NewProxier returns a new Proxier given an iptables Interface instance.
// Because of the iptables logic, it is assumed that there is only a single Proxier active on a machine.
// An error will be returned if iptables fails to update or acquire the initial lock.
// Once a proxier is created, it will keep iptables up to date in the background and
// will not terminate if a particular iptables call fails.
func NewProxier(ipFamily v1.IPFamily,
	ipt utiliptables.Interface,
	sysctl utilsysctl.Interface,
	exec utilexec.Interface,
	syncPeriod time.Duration,
	minSyncPeriod time.Duration,
	masqueradeAll bool,
	localhostNodePorts bool,
	masqueradeBit int,
	localDetector proxyutiliptables.LocalTrafficDetector,
	hostname string,
	nodeIP net.IP,
	recorder events.EventRecorder,
	healthzServer healthcheck.ProxierHealthUpdater,
	nodePortAddressStrings []string,
) (*Proxier, error) {
	// (中略)
	proxier := &Proxier{
		svcPortMap:               make(proxy.ServicePortMap),
		serviceChanges:           proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
		endpointsMap:             make(proxy.EndpointsMap),
		endpointsChanges:         proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil),
		needFullSync:             true,
		syncPeriod:               syncPeriod,
		iptables:                 ipt,
		masqueradeAll:            masqueradeAll,
		masqueradeMark:           masqueradeMark,
		exec:                     exec,
		localDetector:            localDetector,
		hostname:                 hostname,
		nodeIP:                   nodeIP,
		recorder:                 recorder,
		serviceHealthServer:      serviceHealthServer,
		healthzServer:            healthzServer,
		precomputedProbabilities: make([]string, 0, 1001),
		iptablesData:             bytes.NewBuffer(nil),
		existingFilterChainsData: bytes.NewBuffer(nil),
		filterChains:             proxyutil.NewLineBuffer(),
		filterRules:              proxyutil.NewLineBuffer(),
		natChains:                proxyutil.NewLineBuffer(),
		natRules:                 proxyutil.NewLineBuffer(),
		localhostNodePorts:       localhostNodePorts,
		nodePortAddresses:        nodePortAddresses,
		networkInterfacer:        proxyutil.RealNetwork{},
    burstSyncs := 2
	klog.V(2).InfoS("Iptables sync params", "ipFamily", ipt.Protocol(), "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs)
	// We pass syncPeriod to ipt.Monitor, which will call us only if it needs to.
	// We need to pass *some* maxInterval to NewBoundedFrequencyRunner anyway though.
	// time.Hour is arbitrary.
	proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, time.Hour, burstSyncs)
	// (中略)
	return proxier, nil
iptables.NewProxier() が返す Proxier 構造体は proxy.Provider インターフェースを実装しており、同ファイルには ServiceHandler, EndpointSliceHandler インターフェースを実装する各種メソッドや、sync(), syncLoop() メソッド、またそれらの中で呼ばれ実際に iptables を操作する syncProxyRules() メソッドが実装されています。リソースの状態に応じて具体的にどのように iptables を操作するかはこの syncProxyRules() メソッドの中を見ていくことになります。(ちなみにこのメソッドだけで 800 行くらいあります...)
proxier.go




    

// #L759
// This is where all of the iptables-save/restore calls happen.
// The only other iptables rules are those that are setup in iptablesInit()
// This assumes proxier.mu is NOT held
func (proxier *Proxier) syncProxyRules() {
	proxier.mu.Lock()
	defer proxier.mu.Unlock()
    // don't sync rules till we've received services and endpoints
	if !proxier.isInitialized() {
		klog.V(2).InfoS("Not syncing iptables until Services and Endpoints have been received from master")
		return
    (後略)
本記事では iptables プロキシモードの kube-proxy で iptables がどの様に使われるのかを確認するための準備として、kube-proxy, iptables の概要や kube-proxy の実装について簡単に説明しました。次回からは具体的に kube-proxy の中でどの様に iptables が使われているのかを確認していきます。
最後まで読んでいただき、ありがとうございました。
[1] Kubernetesのコンポーネント | Kubernetes

[2] 仮想IPとサービスプロキシー | Kubernetes

[3] iptables | ArchWiki

[4] kubernetes/kubernetes | GitHub