【Troubleshooting】Summary of kube-apiserver troubleshooting techniques (analysis of logs and caching principles)

// k8s.io/utils/trace/trace.go

// LogIfLong is used to dump steps that took longer than its share
// Generally, the input parameter is 500 ms, that is, if the entire request exceeds 500 ms, the Trace log will be printed.
func (t *Trace) LogIfLong(threshold time.Duration) {
    if time.Since(t.startTime) >= threshold {
        // if any step took more than it's share of the total allowed time, it deserves a higher log level
        //The timeout threshold for each step is: total threshold (500ms)/number of steps, that is, the average value
    stepThreshold := threshold / time.Duration(len(t.steps)+1)
        t.logWithStepThreshold(stepThreshold)
    }
}

func (t *Trace) logWithStepThreshold(stepThreshold time.Duration) {
    var buffer bytes.Buffer
    tracenum := rand.Int31()
    endTime := time.Now()

    totalTime := endTime.Sub(t.startTime)
    buffer.WriteString(fmt.Sprintf("Trace[%d]: %q ", tracenum, t.name))
    if len(t.fields) > 0 {
        writeFields(&buffer, t.fields)
        buffer.WriteString(" ")
    }
    buffer.WriteString(fmt.Sprintf("(started: %v) (total time: %v):\n", t.startTime, totalTime))
    lastStepTime := t.startTime
    for _, step := range t.steps {
        stepDuration := step.stepTime.Sub(lastStepTime)
        // If a single step exceeds stepThreshold or the log level is greater than 4, print the log of each Step in Trace
        if stepThreshold == 0 || stepDuration > stepThreshold || klog.V(4) {
            buffer.WriteString(fmt.Sprintf("Trace[%d]: [%v] [%v] ", tracenum, step.stepTime.Sub(t.startTime), stepDuration))
            buffer.WriteString(step.msg)
            if len(step.fields) > 0 {
                buffer.WriteString(" ")
                writeFields(&buffer, step.fields)
            }
            buffer.WriteString("\n")
        }
        lastStepTime = step.stepTime
    }
    stepDuration := endTime.Sub(lastStepTime)
    if stepThreshold == 0 || stepDuration > stepThreshold || klog.V(4) {
        buffer.WriteString(fmt.Sprintf("Trace[%d]: [%v] [%v] END\n", tracenum, endTime.Sub(t.startTime), stepDuration))
    }

    klog.Info(buffer.String())
}

TOKEN=$(kubectl get secrets -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='kube-admin')].data.token}" -n kube-system|base64 --decode)

curl -XPUT -k -H"Authorization: Bearer $TOKEN" https://127.0.0.1:60002/debug/flags/v -d '4'

{
    "kind": "Event",
    "apiVersion": "audit.k8s.io/v1",
    "level": "Request",
    "auditID": "ceffa547-327f-4afe-8d8e-b6e33d4e4e41",
    "stage": "ResponseComplete",
    "requestURI": "/api/v1/namespaces/xxx/configmaps/xxx",
    "verb": "update",
    "user": {
        "username": "user1",
        "groups": [
            "tenant:default"
        ]
    },
    "sourceIPs": [
        "11.11.11.100"
    ],
    // With client fields, it is easy to find the source of the request
    "userAgent": "test-operator/v0.0.0 (linux/amd64) kubernetes/$Format/platform.test_operator",
    "objectRef": {
        "resource": "configmaps",
        "namespace": "xxx",
        "name": "test-operator",
        "uid": "ca529b31-8b62-4163-9974-136246e39024",
        "apiVersion": "v1",
        "resourceVersion": "18892810911"
    },
    "responseStatus": {
        "metadata": {},
        "code": 200
    },
    "requestObject": {
        ...
    }
}

// staging/src/k8s.io/apiserver/pkg/server/options/etcd.go

func NewEtcdOptions(backendConfig *storagebackend.Config) *EtcdOptions {
	options := &EtcdOptions{
		StorageConfig:           *backendConfig,
		DefaultStorageMediaType: "application/json",
		DeleteCollectionWorkers: 1,
		EnableGarbageCollection: true,
		EnableWatchCache:        true,
		// default cache of 100 event changes
		DefaultWatchCacheSize:   100,
	}
	options.StorageConfig.CountMetricPollPeriod = time.Minute
	return options
}

type watchCache struct {
	
	// Maximum size of history window.
	capacity int

	// cache is used a cyclic buffer - its first element (with the smallest
	// resourceVersion) is defined by startIndex, its last element is defined
	// by endIndex (if cache is full it will be startIndex + capacity).
	// Both startIndex and endIndex can be greater than buffer capacity -
	// you should always apply modulo capacity to get an index in cache array.
	cache      []*watchCacheEvent
	
    ...
}

// pkg/registry/cachesize/cachesize.go

// NewHeuristicWatchCacheSizes returns a map of suggested watch cache sizes based on total
// memory.
func NewHeuristicWatchCacheSizes(expectedRAMCapacityMB int) map[schema.GroupResource]int {
	// From our documentation, we officially recommend 120GB machines for
	// 2000 nodes, and we scale from that point. Thus we assume ~60MB of
	// capacity per node.
	clusterSize := expectedRAMCapacityMB / 60

	// We should specify cache size for a given resource only if it
	// is supposed to have non-default value.
	watchCacheSizes := make(map[schema.GroupResource]int)
	watchCacheSizes[schema.GroupResource{Resource: "replicationcontrollers"}] = maxInt(5*clusterSize, 100)
	watchCacheSizes[schema.GroupResource{Resource: "endpoints"}] = maxInt(10*clusterSize, 1000)
	watchCacheSizes[schema.GroupResource{Resource: "endpointslices", Group: "discovery.k8s.io"}] = maxInt(10*clusterSize, 1000)
	watchCacheSizes[schema.GroupResource{Resource: "nodes"}] = maxInt(5*clusterSize, 1000)
	watchCacheSizes[schema.GroupResource{Resource: "pods"}] = maxInt(50*clusterSize, 1000)
	watchCacheSizes[schema.GroupResource{Resource: "services"}] = maxInt(5*clusterSize, 1000)
	// The cacheSize of event is 0, do not go to the cache, directly query etcd
	watchCacheSizes[schema.GroupResource{Resource: "events"}] = 0
	watchCacheSizes[schema.GroupResource{Resource: "events", Group: "events.k8s.io"}] = 0
	watchCacheSizes[schema.GroupResource{Resource: "apiservices", Group: "apiregistration.k8s.io"}] = maxInt(5*clusterSize, 1000)
	watchCacheSizes[schema.GroupResource{Resource: "leases", Group: "coordination.k8s.io"}] = maxInt(5*clusterSize, 1000)
	return watchCacheSizes
}

// staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion/types.go

// When specified with a watch call, shows changes that occur after that particular version of a resource.
// Defaults to changes from the beginning of history.
// When specified for list:
// - if unset, then the result is returned from remote storage based on quorum-read flag;
// - if it's 0, then we simply return what we currently have in cache, no guarantee;
// - if set to non zero, then the result is at least as fresh as given rv.
ResourceVersion string

// k8s.io/apiserver/pkg/storage/cacher/cacher.go

// List implements storage.Interface.
func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate, listObj runtime.Object) error {
	pagingEnabled := utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
	hasContinuation := pagingEnabled && len(pred.Continue) > 0
	// Conditions for enabling limit paging: enable the APIListChunking function, request parameters with limit, and resourceVersion is not 0
	hasLimit := pagingEnabled && pred.Limit > 0 && resourceVersion != "0"
	// Directly request the backend etcd when the following conditions are met:
	if resourceVersion == "" || hasContinuation || hasLimit {
		// If resourceVersion is not specified, serve it from underlying
		// storage (for backward compatibility). If a continuation is
		// requested, serve it from the underlying storage as well.
		// Limits are only sent to storage when resourceVersion is non-zero
		// since the watch cache isn't able to perform continuations, and
		// limits are ignored when resource version is zero.
		return c.storage.List(ctx, key, resourceVersion, pred, listObj)
	}

	// If resourceVersion is specified, serve it from cache.
	// It's guaranteed that the returned value is at least that
	// fresh as the given resourceVersion.
	listRV, err := c.versioner.ParseResourceVersion(resourceVersion)
	if err != nil {
		return err
	}

	// 缓存还没有 ready，直接请求后端 etcd
	if listRV == 0 && !c.ready.check() {
		// If Cacher is not yet initialized and we don't require any specific
		// minimal resource version, simply forward the request to storage.
		return c.storage.List(ctx, key, resourceVersion, pred, listObj)
	}

	trace := utiltrace.New("cacher list", utiltrace.Field{"type", c.objectType.String()})
	defer trace.LogIfLong(500 * time.Millisecond)

	c.ready.wait()
	trace.Step("Ready")

	// List elements with at least 'listRV' from cache.
	listPtr, err := meta.GetItemsPtr(listObj)
	if err != nil {
		return err
	}
	listVal, err := conversion.EnforcePtr(listPtr)
	if err != nil {
		return err
	}
	if listVal.Kind() != reflect.Slice {
		return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind())
	}
	filter := filterWithAttrsFunction(key, pred)

	objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV, pred.MatcherIndex(), trace)
	if err != nil {
		return err
	}
	trace.Step("Listed items from cache", utiltrace.Field{"count", len(objs)})
	if len(objs) > listVal.Cap() && pred.Label.Empty() && pred.Field.Empty() {
		// Resize the slice appropriately, since we already know that none
		// of the elements will be filtered out.
		listVal.Set(reflect.MakeSlice(reflect.SliceOf(c.objectType.Elem()), 0, len(objs)))
		trace.Step("Resized result")
	}
	for _, obj := range objs {
		elem, ok := obj.(*storeElement)
		if !ok {
			return fmt.Errorf("non *storeElement returned from storage: %v", obj)
		}
		if filter(elem.Key, elem.Labels, elem.Fields) {
			listVal.Set(reflect.Append(listVal, reflect.ValueOf(elem.Object).Elem()))
		}
	}
	trace.Step("Filtered items", utiltrace.Field{"count", listVal.Len()})
	if c.versioner != nil {
		if err := c.versioner.UpdateList(listObj, readResourceVersion, "", nil); err != nil {
			return err
		}
	}
	return nil
}

GET /api/v1/namespaces/test/pods?watch=1&resourceVersion=10245&allowWatchBookmarks=true
---
200 OK
Transfer-Encoding: chunked
Content-Type: application/json

{
  "type": "ADDED",
  "object": {"kind": "Pod", "apiVersion": "v1", "metadata": {"resourceVersion": "10596", ...}, ...}
}
...
{
  "type": "BOOKMARK",
  "object": {"kind": "Pod", "apiVersion": "v1", "metadata": {"resourceVersion": "12746"} }
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case err := <-errc:
			return err
		case event, ok := <-w.ResultChan():
			...
			switch event.Type {
			case watch.Added:
				...
			case watch.Modified:
				...
			case watch.Deleted:
				...
			case watch.Bookmark:
				// 接收到 bookmark 事件不做任何处理，只更新 watcher 的 lastSyncResourceVersion 为最新的 ResourceVersion
				// A `Bookmark` means watch has synced here, just update the resourceVersion
			default:
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = newResourceVersion
			r.setLastSyncResourceVersion(newResourceVersion)
			eventCount++
		}
	}
    ...
	return nil
}

// update the latest resourceversion
func (r *Reflector) setLastSyncResourceVersion(v string) {
    r.lastSyncResourceVersionMutex.Lock()
    defer r.lastSyncResourceVersionMutex.Unlock()
    r.lastSyncResourceVersion = v
}

// When relisting is required, start watching from the latest ResourceVersion. There is no need to start the full list from ResourceVersion=0.
func (r *Reflector) relistResourceVersion() string {
    r.lastSyncResourceVersionMutex.RLock()
    defer r.lastSyncResourceVersionMutex.RUnlock()

    if r.isLastSyncResourceVersionUnavailable {
        // Since this reflector makes paginated list requests, and all paginated list requests skip the watch cache
        // if the lastSyncResourceVersion is unavailable, we set ResourceVersion="" and list again to re-establish reflector
        // to the latest available ResourceVersion, using a consistent read from etcd.
        return ""
    }
    if r.lastSyncResourceVersion == "" {
        // For performance reasons, initial list performed by reflector uses "0" as resource version to allow it to
        // be served from the watch cache if it is enabled.
        return "0"
    }
    return r.lastSyncResourceVersion
}

// staging/src/k8s.io/client-go/tools/cache/reflector.go
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
	var resourceVersion string

	// The ResourceVersion of the first list here defaults to 0
	options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}

	if err := func() error {
		...
		go func() {
			...
			// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
			// list request will return the full response.
			pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
		}
		...
    }
    ...
}

// staging/src/k8s.io/client-go/tools/pager/pager.go

// default is limit 500 paging parameter
const defaultPageSize = 500

func New(fn ListPageFunc) *ListPager {
	return &ListPager{
		PageSize:          defaultPageSize,
		PageFn:            fn,
		FullListIfExpired: true,
		PageBufferSize:    defaultPageBufferSize,
	}
}

GET https://kube-apiserver/api/v1/pods?fieldSelector=spec.nodeName=node1,status.phase!=Failed,status.phase!=Succeeded

// pkg/registry/core/pod/strategy.go

// Indexers returns the indexers for pod storage.
func Indexers() *cache.Indexers {
	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.SelectorIndex) {
		return &cache.Indexers{
			storage.FieldIndex("spec.nodeName"): NodeNameIndexFunc,
		}
	}
	return nil
}