はじめに

本記事はサイバーエージェント24卒内定者 Advent Calendar 2023 6日目の記事です。

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

kube-proxy とは

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

(引用: 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 からの通信を遮断する場合は以下の様なコマンドで実現できます。

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