5. 大规模集群优化  

FROM pause:3.15
RUN dd if=/dev/urandom of=/data bs=1M count=256
CMD ["/pause"]

* soft     nproc          2097152
* hard     nproc          2097152
* soft     nofile         2097152
* hard     nofile         2097152
root soft     nproc          2097152
root hard     nproc          2097152
root soft     nofile         2097152
root hard     nofile         2097152

ip_vs
ip_vs_rr
ip_vs_wrr
ip_vs_sh
nf_conntrack
ip_tables
iptable_filter
ipt_state
br_netfilter

# load nf_conntrack_ipv4 if it exists
nf_conntrack_ipv4_exist=$(modinfo nf_conntrack_ipv4 2>&1|grep filename|wc -l)
if [ $nf_conntrack_ipv4_exist -gt 0 ];then
  echo -ne "nf_conntrack_ipv4
  " >> /etc/modules-load.d/k8s-optimize.conf
  modprobe nf_conntrack_ipv4
fi

# file descriptor
fs.inotify.max_user_instances=16384
fs.inotify.max_user_watches=2097152
fs.file-max=2097152
fs.nr_open=2097152
# nf_conntrack
# max:buckets -> 4:1
net.netfilter.nf_conntrack_max=2097152
net.netfilter.nf_conntrack_buckets=524288
## linux/k8s: 600 cks: 120
net.netfilter.nf_conntrack_generic_timeout=120
# nf_conntrack_tcp
## linux: 0 cks: 1
net.netfilter.nf_conntrack_tcp_be_liberal=1
net.netfilter.nf_conntrack_tcp_loose=1
net.netfilter.nf_conntrack_tcp_max_retrans=3
net.netfilter.nf_conntrack_tcp_timeout_close=10
## linux: 60 k8s: 3600 cks: 30
net.netfilter.nf_conntrack_tcp_timeout_close_wait=30
## linux: 43200 k8s: 86400, cks: 1800
net.netfilter.nf_conntrack_tcp_timeout_established=1800
## linux/k8s: 120 cks: 30
net.netfilter.nf_conntrack_tcp_timeout_fin_wait=30
net.netfilter.nf_conntrack_tcp_timeout_last_ack=30
net.netfilter.nf_conntrack_tcp_timeout_max_retrans=300
## linux/k8s: 60 cks: 30
net.netfilter.nf_conntrack_tcp_timeout_syn_recv=30
## linux/k8s: 120 cks: 30
net.netfilter.nf_conntrack_tcp_timeout_syn_sent=30
## linux/k8s: 120 cks: 30
net.netfilter.nf_conntrack_tcp_timeout_time_wait=30
net.netfilter.nf_conntrack_tcp_timeout_unacknowledged=300
# tcp-ipv4
## linux/k8s: 2 cks: 1
net.ipv4.tcp_tw_reuse=1
net.ipv4.tcp_timestamps=1
## tcp_keepalive
net.ipv4.tcp_keepalive_time=900
net.ipv4.tcp_keepalive_intvl=75
net.ipv4.tcp_keepalive_probes=9
# arp-ipv4
## linux/k8s: 128,512,1024 cks: 8192,16384,32768
net.ipv4.neigh.default.gc_thresh1=8192
net.ipv4.neigh.default.gc_thresh2=16384
net.ipv4.neigh.default.gc_thresh3=32768
# arp-ipv6
## linux/k8s: 128,512,1024 cks: 8192,16384,32768
net.ipv6.neigh.default.gc_thresh1=8192
net.ipv6.neigh.default.gc_thresh2=16384
net.ipv6.neigh.default.gc_thresh3=32768
# forward
net.ipv4.ip_forward=1
net.ipv6.conf.all.forwarding=1

apiVersion: apps/v1
kind: Deployment
metadata:
  name: pending-pods
  namespace: user
spec:
  replicas: 100000  # 设置所需的 Pod 数量
  selector:
    matchLabels:
      app: pending-app
  template:
    metadata:
      labels:
        app: pending-app
    spec:
      nodeSelector:
        disktype: ssd  # 选择节点的标签
      containers:
      - name: pending-container
        image: image.wbuntu.com/library/nginx:1.20-alpine  # 使用轻量级镜像
        resources:
          requests:
            memory: "2Gi"  # 请求较高的资源
            cpu: "1000m"   # 请求较高的 CPU
          limits:
            memory: "2Gi"
            cpu: "1000m"

module github.com/wbuntu/podcache

go 1.23.0

require (
	k8s.io/api v0.31.0
	k8s.io/apimachinery v0.31.0
	k8s.io/client-go v0.31.0
	sigs.k8s.io/controller-runtime v0.19.0
)

require (
	github.com/beorn7/perks v1.0.1 // indirect
	github.com/cespare/xxhash/v2 v2.3.0 // indirect
	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
	github.com/evanphx/json-patch/v5 v5.9.0 // indirect
	github.com/fsnotify/fsnotify v1.7.0 // indirect
	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
	github.com/go-logr/logr v1.4.2 // indirect
	github.com/go-logr/zapr v1.3.0 // indirect
	github.com/go-openapi/jsonpointer v0.19.6 // indirect
	github.com/go-openapi/jsonreference v0.20.2 // indirect
	github.com/go-openapi/swag v0.22.4 // indirect
	github.com/gogo/protobuf v1.3.2 // indirect
	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
	github.com/golang/protobuf v1.5.4 // indirect
	github.com/google/gnostic-models v0.6.8 // indirect
	github.com/google/go-cmp v0.6.0 // indirect
	github.com/google/gofuzz v1.2.0 // indirect
	github.com/google/uuid v1.6.0 // indirect
	github.com/imdario/mergo v0.3.6 // indirect
	github.com/josharian/intern v1.0.0 // indirect
	github.com/json-iterator/go v1.1.12 // indirect
	github.com/mailru/easyjson v0.7.7 // indirect
	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
	github.com/modern-go/reflect2 v1.0.2 // indirect
	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
	github.com/pkg/errors v0.9.1 // indirect
	github.com/prometheus/client_golang v1.19.1 // indirect
	github.com/prometheus/client_model v0.6.1 // indirect
	github.com/prometheus/common v0.55.0 // indirect
	github.com/prometheus/procfs v0.15.1 // indirect
	github.com/spf13/pflag v1.0.5 // indirect
	github.com/x448/float16 v0.8.4 // indirect
	go.uber.org/multierr v1.11.0 // indirect
	go.uber.org/zap v1.26.0 // indirect
	golang.org/x/exp v0.0.0-20230515195305-f3d0a9c9a5cc // indirect
	golang.org/x/net v0.26.0 // indirect
	golang.org/x/oauth2 v0.21.0 // indirect
	golang.org/x/sys v0.21.0 // indirect
	golang.org/x/term v0.21.0 // indirect
	golang.org/x/text v0.16.0 // indirect
	golang.org/x/time v0.3.0 // indirect
	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
	google.golang.org/protobuf v1.34.2 // indirect
	gopkg.in/inf.v0 v0.9.1 // indirect
	gopkg.in/yaml.v2 v2.4.0 // indirect
	gopkg.in/yaml.v3 v3.0.1 // indirect
	k8s.io/apiextensions-apiserver v0.31.0 // indirect
	k8s.io/klog/v2 v2.130.1 // indirect
	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect
	k8s.io/utils v0.0.0-20240711033017-18e509b52bc8 // indirect
	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
	sigs.k8s.io/yaml v1.4.0 // indirect
)

package main

import (
	"context"
	"encoding/json"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/clientcmd/api"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
)

type reconciler struct {
	client.Client
	scheme *runtime.Scheme
	cache  *Cache
	store  cache.Cache
}

func (r *reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx).WithValues("pod", req.NamespacedName)

	var pod corev1.Pod
	if err := r.Get(ctx, req.NamespacedName, &pod); err != nil {
		if !apierrors.IsNotFound(err) {
			log.Error(err, "unable to get pod")
			return ctrl.Result{}, err
		}
		r.cache.Delete(req.Namespace, req.Name)
		log.V(1).Info("Delete pod")
	} else {
		item := Item{
			Namespace: pod.Namespace,
			Name:      pod.Name,
			Timestamp: pod.CreationTimestamp.Time,
		}
		r.cache.Add(item)
		log.V(1).Info("Add pod")
	}
	return ctrl.Result{}, nil
}

// PodResponse 用于响应 Pod 列表
type PodResponse struct {
	Pods  []corev1.Pod `json:"pods"`
	Total int          `json:"total"`
}

// httpHandler 处理 HTTP 请求
func (r *reconciler) httpHandler(w http.ResponseWriter, req *http.Request) {
	namespace := req.URL.Query().Get("namespace")
	pageStr := req.URL.Query().Get("page")
	pageSizeStr := req.URL.Query().Get("pageSize")
	search := req.URL.Query().Get("search")

	page := 1
	pageSize := 10

	if p, err := strconv.Atoi(pageStr); err == nil {
		page = p
	}
	if ps, err := strconv.Atoi(pageSizeStr); err == nil {
		pageSize = ps
	}

	var items []*Item
	var total int
	if namespace != "" {
		items, total = r.cache.GetPagedByNamespace(namespace, search, page, pageSize)
	} else {
		items, total = r.cache.GetPaged(search, page, pageSize)
	}

	pods := make([]corev1.Pod, 0, len(items))
	for _, item := range items {
		var pod corev1.Pod
		err := r.store.Get(context.TODO(), client.ObjectKey{Namespace: item.Namespace, Name: item.Name}, &pod)
		if err == nil {
			pods = append(pods, pod)
		}
	}

	response := PodResponse{
		Pods:  pods,
		Total: total,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

func main() {
	ctrl.SetLogger(zap.New())
	setupLog := ctrl.Log.WithName("setup")
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{})
	if err != nil {
		setupLog.Error(err, "unable to start manager")
		os.Exit(1)
	}

	// in a real controller, we'd create a new scheme for this
	if err := api.AddToScheme(mgr.GetScheme()); err != nil {
		setupLog.Error(err, "unable to add scheme")
		os.Exit(1)
	}
	reconciler := &reconciler{
		Client: mgr.GetClient(),
		scheme: mgr.GetScheme(),
		cache:  NewCache(),
		store:  mgr.GetCache(),
	}
	if err := ctrl.NewControllerManagedBy(mgr).For(&corev1.Pod{}).Complete(reconciler); err != nil {
		setupLog.Error(err, "unable to create controller")
		os.Exit(1)
	}

	// 添加 HTTP 处理程序
	http.HandleFunc("/pods", reconciler.httpHandler)

	go func() {
		setupLog.Info("starting http server")
		if err := http.ListenAndServe(":9090", nil); err != nil {
			setupLog.Error(err, "unable to start HTTP server")
			os.Exit(1)
		}
	}()

	setupLog.Info("starting manager")
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}
}

// Item 表示缓存中的项目
type Item struct {
	Namespace string
	Name      string
	Timestamp time.Time
}

// Cache 用于存储 Pod 项目
type Cache struct {
	mu    sync.RWMutex
	items map[string]struct{} // 用于去重
	order []*Item             // 用于保存顺序
}

// NewCache 创建新的 Cache 实例
func NewCache() *Cache {
	return &Cache{
		items: make(map[string]struct{}),
		order: make([]*Item, 0),
	}
}

// Add 添加项到缓存
func (c *Cache) Add(item Item) {
	c.mu.Lock()
	defer c.mu.Unlock()

	key := item.Namespace + "/" + item.Name
	if _, exists := c.items[key]; !exists {
		c.items[key] = struct{}{}
		c.order = append(c.order, &item) // 添加到顺序列表
	}
}

// Delete 从缓存中删除项
func (c *Cache) Delete(namespace, name string) {
	c.mu.Lock()
	defer c.mu.Unlock()

	key := namespace + "/" + name
	if _, exists := c.items[key]; exists {
		delete(c.items, key)

		// 从顺序列表中删除
		for i, item := range c.order {
			if item.Namespace == namespace && item.Name == name {
				c.order = append(c.order[:i], c.order[i+1:]...) // 删除项
				break
			}
		}
	}
}

// GetPaged 获取分页数据
func (c *Cache) GetPaged(search string, page int, pageSize int) ([]*Item, int) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	var filteredItems []*Item
	for _, item := range c.order {
		if contains(item.Name, search) {
			filteredItems = append(filteredItems, item)
		}
	}

	total := len(filteredItems)
	start := (page - 1) * pageSize
	end := start + pageSize

	if start > total {
		return []*Item{}, total
	}
	if end > total {
		end = total
	}

	return filteredItems[start:end], total
}

// GetPagedByNamespace 根据命名空间获取分页数据
func (c *Cache) GetPagedByNamespace(namespace string, search string, page int, pageSize int) ([]*Item, int) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	var filteredItems []*Item
	for _, item := range c.order {
		if item.Namespace == namespace && contains(item.Name, search) {
			filteredItems = append(filteredItems, item)
		}
	}

	total := len(filteredItems)
	start := (page - 1) * pageSize
	end := start + pageSize

	if start > total {
		return []*Item{}, total
	}
	if end > total {
		end = total
	}

	return filteredItems[start:end], total
}

// contains 检查字符串是否包含子字符串
func contains(str, substr string) bool {
	return substr == "" || strings.Contains(str, substr)
}

➜  ~ etcdctl endpoint status -w table
+----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+
|       ENDPOINT       |        ID        | VERSION | DB SIZE | IS LEADER | IS LEARNER | RAFT TERM | RAFT INDEX | RAFT APPLIED INDEX | ERRORS |
+----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+
| 192.168.123.101:2379 |  d32a4b16a0f9191 |  3.5.15 |  832 MB |      true |      false |         3 |    3712326 |            3712325 |        |
| 192.168.123.102:2379 | 47c5ecfae8fccdb1 |  3.5.15 |  832 MB |     false |      false |         3 |    3712326 |            3712326 |        |
| 192.168.123.103:2379 |  96224218d3d7de4 |  3.5.15 |  832 MB |     false |      false |         3 |    3712326 |            3712326 |        |
+----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+

➜  ~ /usr/bin/time -v kubectl get pod -A > /dev/null
	Command being timed: "kubectl get pod -A"
	User time (seconds): 26.44
	System time (seconds): 1.80
	Percent of CPU this job got: 55%
	Elapsed (wall clock) time (h:mm:ss or m:ss): 0:50.98
	Average shared text size (kbytes): 0
	Average unshared data size (kbytes): 0
	Average stack size (kbytes): 0
	Average total size (kbytes): 0
	Maximum resident set size (kbytes): 2438748
	Average resident set size (kbytes): 0
	Major (requiring I/O) page faults: 0
	Minor (reclaiming a frame) page faults: 706168
	Voluntary context switches: 21425
	Involuntary context switches: 11041
	Swaps: 0
	File system inputs: 0
	File system outputs: 0
	Socket messages sent: 0
	Socket messages received: 0
	Signals delivered: 0
	Page size (bytes): 4096
	Exit status: 0

➜  ~ date && curl "http://192.168.123.3:9090/pods?page=1&pageSize=100007" -o pod.json && date
Wed Sep  4 09:59:39 PM CST 2024
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  317M    0  317M    0     0  60.6M      0 --:--:--  0:00:05 --:--:-- 78.8M
Wed Sep  4 09:59:45 PM CST 2024

➜  ~ time curl -s "http://192.168.123.3:9090/pods?page=10000&pageSize=10" -o /dev/null
curl -s "http://192.168.123.3:9090/pods?page=10000&pageSize=10" -o /dev/null  0.01s user 0.01s system 77% cpu 0.014 total