cache.go

Documentation: k8s.io/kube-openapi/pkg/cached

     1  /*
     2  Copyright 2022 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  // Package cached provides a cache mechanism based on etags to lazily
    18  // build, and/or cache results from expensive operation such that those
    19  // operations are not repeated unnecessarily. The operations can be
    20  // created as a tree, and replaced dynamically as needed.
    21  //
    22  // All the operations in this module are thread-safe.
    23  //
    24  // # Dependencies and types of caches
    25  //
    26  // This package uses a source/transform/sink model of caches to build
    27  // the dependency tree, and can be used as follows:
    28  //   - [Func]: A source cache that recomputes the content every time.
    29  //   - [Once]: A source cache that always produces the
    30  //     same content, it is only called once.
    31  //   - [Transform]: A cache that transforms data from one format to
    32  //     another. It's only refreshed when the source changes.
    33  //   - [Merge]: A cache that aggregates multiple caches in a map into one.
    34  //     It's only refreshed when the source changes.
    35  //   - [MergeList]: A cache that aggregates multiple caches in a list into one.
    36  //     It's only refreshed when the source changes.
    37  //   - [Atomic]: A cache adapter that atomically replaces the source with a new one.
    38  //   - [LastSuccess]: A cache adapter that caches the last successful and returns
    39  //     it if the next call fails. It extends [Atomic].
    40  //
    41  // # Etags
    42  //
    43  // Etags in this library is a cache version identifier. It doesn't
    44  // necessarily strictly match to the semantics of http `etags`, but are
    45  // somewhat inspired from it and function with the same principles.
    46  // Hashing the content is a good way to guarantee that your function is
    47  // never going to be called spuriously. In Kubernetes world, this could
    48  // be a `resourceVersion`, this can be an actual etag, a hash, a UUID
    49  // (if the cache always changes), or even a made-up string when the
    50  // content of the cache never changes.
    51  package cached
    52  
    53  import (
    54  	"fmt"
    55  	"sync"
    56  	"sync/atomic"
    57  )
    58  
    59  // Value is wrapping a value behind a getter for lazy evaluation.
    60  type Value[T any] interface {
    61  	Get() (value T, etag string, err error)
    62  }
    63  
    64  // Result is wrapping T and error into a struct for cases where a tuple is more
    65  // convenient or necessary in Golang.
    66  type Result[T any] struct {
    67  	Value T
    68  	Etag  string
    69  	Err   error
    70  }
    71  
    72  func (r Result[T]) Get() (T, string, error) {
    73  	return r.Value, r.Etag, r.Err
    74  }
    75  
    76  // Func wraps a (thread-safe) function as a Value[T].
    77  func Func[T any](fn func() (T, string, error)) Value[T] {
    78  	return valueFunc[T](fn)
    79  }
    80  
    81  type valueFunc[T any] func() (T, string, error)
    82  
    83  func (c valueFunc[T]) Get() (T, string, error) {
    84  	return c()
    85  }
    86  
    87  // Static returns constant values.
    88  func Static[T any](value T, etag string) Value[T] {
    89  	return Result[T]{Value: value, Etag: etag}
    90  }
    91  
    92  // Merge merges a of cached values. The merge function only gets called if any of
    93  // the dependency has changed.
    94  //
    95  // If any of the dependency returned an error before, or any of the
    96  // dependency returned an error this time, or if the mergeFn failed
    97  // before, then the function is run again.
    98  //
    99  // Note that this assumes there is no "partial" merge, the merge
   100  // function will remerge all the dependencies together everytime. Since
   101  // the list of dependencies is constant, there is no way to save some
   102  // partial merge information either.
   103  //
   104  // Also note that Golang map iteration is not stable. If the mergeFn
   105  // depends on the order iteration to be stable, it will need to
   106  // implement its own sorting or iteration order.
   107  func Merge[K comparable, T, V any](mergeFn func(results map[K]Result[T]) (V, string, error), caches map[K]Value[T]) Value[V] {
   108  	list := make([]Value[T], 0, len(caches))
   109  
   110  	// map from index to key
   111  	indexes := make(map[int]K, len(caches))
   112  	i := 0
   113  	for k := range caches {
   114  		list = append(list, caches[k])
   115  		indexes[i] = k
   116  		i++
   117  	}
   118  
   119  	return MergeList(func(results []Result[T]) (V, string, error) {
   120  		if len(results) != len(indexes) {
   121  			panic(fmt.Errorf("invalid result length %d, expected %d", len(results), len(indexes)))
   122  		}
   123  		m := make(map[K]Result[T], len(results))
   124  		for i := range results {
   125  			m[indexes[i]] = results[i]
   126  		}
   127  		return mergeFn(m)
   128  	}, list)
   129  }
   130  
   131  // MergeList merges a list of cached values. The function only gets called if
   132  // any of the dependency has changed.
   133  //
   134  // The benefit of ListMerger over the basic Merger is that caches are
   135  // stored in an ordered list so the order of the cache will be
   136  // preserved in the order of the results passed to the mergeFn.
   137  //
   138  // If any of the dependency returned an error before, or any of the
   139  // dependency returned an error this time, or if the mergeFn failed
   140  // before, then the function is reran.
   141  //
   142  // Note that this assumes there is no "partial" merge, the merge
   143  // function will remerge all the dependencies together everytime. Since
   144  // the list of dependencies is constant, there is no way to save some
   145  // partial merge information either.
   146  func MergeList[T, V any](mergeFn func(results []Result[T]) (V, string, error), delegates []Value[T]) Value[V] {
   147  	return &listMerger[T, V]{
   148  		mergeFn:   mergeFn,
   149  		delegates: delegates,
   150  	}
   151  }
   152  
   153  type listMerger[T, V any] struct {
   154  	lock      sync.Mutex
   155  	mergeFn   func([]Result[T]) (V, string, error)
   156  	delegates []Value[T]
   157  	cache     []Result[T]
   158  	result    Result[V]
   159  }
   160  
   161  func (c *listMerger[T, V]) prepareResultsLocked() []Result[T] {
   162  	cacheResults := make([]Result[T], len(c.delegates))
   163  	ch := make(chan struct {
   164  		int
   165  		Result[T]
   166  	}, len(c.delegates))
   167  	for i := range c.delegates {
   168  		go func(index int) {
   169  			value, etag, err := c.delegates[index].Get()
   170  			ch <- struct {
   171  				int
   172  				Result[T]
   173  			}{index, Result[T]{Value: value, Etag: etag, Err: err}}
   174  		}(i)
   175  	}
   176  	for i := 0; i < len(c.delegates); i++ {
   177  		res := <-ch
   178  		cacheResults[res.int] = res.Result
   179  	}
   180  	return cacheResults
   181  }
   182  
   183  func (c *listMerger[T, V]) needsRunningLocked(results []Result[T]) bool {
   184  	if c.cache == nil {
   185  		return true
   186  	}
   187  	if c.result.Err != nil {
   188  		return true
   189  	}
   190  	if len(results) != len(c.cache) {
   191  		panic(fmt.Errorf("invalid number of results: %v (expected %v)", len(results), len(c.cache)))
   192  	}
   193  	for i, oldResult := range c.cache {
   194  		newResult := results[i]
   195  		if newResult.Etag != oldResult.Etag || newResult.Err != nil || oldResult.Err != nil {
   196  			return true
   197  		}
   198  	}
   199  	return false
   200  }
   201  
   202  func (c *listMerger[T, V]) Get() (V, string, error) {
   203  	c.lock.Lock()
   204  	defer c.lock.Unlock()
   205  	cacheResults := c.prepareResultsLocked()
   206  	if c.needsRunningLocked(cacheResults) {
   207  		c.cache = cacheResults
   208  		c.result.Value, c.result.Etag, c.result.Err = c.mergeFn(c.cache)
   209  	}
   210  	return c.result.Value, c.result.Etag, c.result.Err
   211  }
   212  
   213  // Transform the result of another cached value. The transformFn will only be called
   214  // if the source has updated, otherwise, the result will be returned.
   215  //
   216  // If the dependency returned an error before, or it returns an error
   217  // this time, or if the transformerFn failed before, the function is
   218  // reran.
   219  func Transform[T, V any](transformerFn func(T, string, error) (V, string, error), source Value[T]) Value[V] {
   220  	return MergeList(func(delegates []Result[T]) (V, string, error) {
   221  		if len(delegates) != 1 {
   222  			panic(fmt.Errorf("invalid cache for transformer cache: %v", delegates))
   223  		}
   224  		return transformerFn(delegates[0].Value, delegates[0].Etag, delegates[0].Err)
   225  	}, []Value[T]{source})
   226  }
   227  
   228  // Once calls Value[T].Get() lazily and only once, even in case of an error result.
   229  func Once[T any](d Value[T]) Value[T] {
   230  	return &once[T]{
   231  		data: d,
   232  	}
   233  }
   234  
   235  type once[T any] struct {
   236  	once   sync.Once
   237  	data   Value[T]
   238  	result Result[T]
   239  }
   240  
   241  func (c *once[T]) Get() (T, string, error) {
   242  	c.once.Do(func() {
   243  		c.result.Value, c.result.Etag, c.result.Err = c.data.Get()
   244  	})
   245  	return c.result.Value, c.result.Etag, c.result.Err
   246  }
   247  
   248  // Replaceable extends the Value[T] interface with the ability to change the
   249  // underlying Value[T] after construction.
   250  type Replaceable[T any] interface {
   251  	Value[T]
   252  	Store(Value[T])
   253  }
   254  
   255  // Atomic wraps a Value[T] as an atomic value that can be replaced. It implements
   256  // Replaceable[T].
   257  type Atomic[T any] struct {
   258  	value atomic.Pointer[Value[T]]
   259  }
   260  
   261  var _ Replaceable[[]byte] = &Atomic[[]byte]{}
   262  
   263  func (x *Atomic[T]) Store(val Value[T])      { x.value.Store(&val) }
   264  func (x *Atomic[T]) Get() (T, string, error) { return (*x.value.Load()).Get() }
   265  
   266  // LastSuccess calls Value[T].Get(), but hides errors by returning the last
   267  // success if there has been any.
   268  type LastSuccess[T any] struct {
   269  	Atomic[T]
   270  	success atomic.Pointer[Result[T]]
   271  }
   272  
   273  var _ Replaceable[[]byte] = &LastSuccess[[]byte]{}
   274  
   275  func (c *LastSuccess[T]) Get() (T, string, error) {
   276  	success := c.success.Load()
   277  	value, etag, err := c.Atomic.Get()
   278  	if err == nil {
   279  		if success == nil {
   280  			c.success.CompareAndSwap(nil, &Result[T]{Value: value, Etag: etag, Err: err})
   281  		}
   282  		return value, etag, err
   283  	}
   284  
   285  	if success != nil {
   286  		return success.Value, success.Etag, success.Err
   287  	}
   288  
   289  	return value, etag, err
   290  }
   291
View as plain text