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

## はじめに

本記事は [サイバーエージェント24卒内定者 Advent Calendar 2023](https://qiita.com/advent-calendar/2023/ca24engineer) 6日目の記事です。

みなさんご存知のコンテナプラットフォーム Kubernetes は様々コンポーネントが組み合わさって構成されており、その中でもワーカーノードのネットワークを制御する 「**kube-proxy**」 というコンポーネントがあります。kube-proxy はワーカーノード内のトラフィックを制御する方式としていくつかの動作モードを持っていますが、今回はその中でも Linux のネットワーク機能である 「**iptables**」 を使用する動作モードの挙動を追ってみます。

## kube-proxy とは

kube-proxy は各ワーカーノードで動くプロセスであり、Pod への通信を可能にするための Service リソース等の作成に応じてワーカーノード内のネットワーク設定を行います。

![Screenshot 2023-12-03 at 12.06.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534248/ca226d64-aa8f-5002-3a44-a20068731e6d.png)
(引用: https://kubernetes.io/ja/docs/concepts/overview/components/)

ネットワーク設定をする具体的な方法としては「**user-space プロキシーモード**」、「**iptables プロキシーモード**」、「**IPVS プロキシーモード**」の3つがありますが、今回は iptables プロキシーモード を利用する場合の実装を追ってみます。[[2]](##参考文献)

## iptables とは

iptables は Linux のネットワーク機能であり、コマンドを通してパケットフィルタリングやルーティング等をカーネルのレイヤで制御することができます。iptables の実装は **Netfilter** という Linux のモジュールであり、実際には iptables は Netfilter を操作するためのインターフェースとなっています。[[3]](##参考文献)

iptables の使い方について簡単に説明すると、例えば特定のIPアドレス `192.0.2.1` からの通信を遮断する場合は以下の様なコマンドで実現できます。

```bash
iptables -A INPUT -s 192.0.2.1 -j DROP
```

また、SSH 通信のみを許可する場合は以下の様になります。

```bash
iptables -A INPUT -p tcp -m tcp --dport 22 -j ACCEPT
```

また、iptables はルールによってパケットの中身を変更することもでき、例えば特定のパケットの宛先を変更する場合 (Destination-NAT) は以下の様なルールを設定します。

```bash
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 の実装の概要について確認しておきましょう。図にすると以下の様になります。

![Screenshot 2023-12-03 at 15.43.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534248/c6f781fa-ff9b-d1cf-b560-bcd4423cb8f5.png)

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](https://kubernetes.io/ja/docs/concepts/overview/components/)
[2] [仮想IPとサービスプロキシー | Kubernetes](https://kubernetes.io/ja/docs/concepts/services-networking/service/#virtual-ips-and-service-proxies)
[3] [iptables | ArchWiki](https://wiki.archlinux.jp/index.php/Iptables)
[4] [kubernetes/kubernetes | GitHub](https://github.com/kubernetes/kubernetes)